PySpark + Google 云存储（wholeTextFiles）
PySpark + Google Cloud Storage (wholeTextFiles)
我正在尝试使用 PySpark (Google Dataproc) 解析大约 100 万个 HTML 文件，并将相关字段写入压缩文件。每个 HTML 文件大约 200KB。因此，所有数据约为 200GB。
如果我使用数据的一个子集，下面的代码工作正常，但运行数小时然后在整个数据集上 运行 时崩溃。此外，未使用工作节点 (<5% CPU)，所以我知道存在一些问题。
我认为系统在从 GCS 摄取数据时出现问题。有一个更好的方法吗？另外，当我以这种方式使用 wholeTextFiles 时，主人是否会尝试下载所有文件然后将它们发送给执行者，还是让执行者下载它们？
def my_func(keyval):
keyval = (file_name, file_str)
return parser(file_str).__dict__
data = sc.wholeTextFiles("gs://data/*")
output = data.map(my_func)
output.saveAsTextFile("gs://results/a")
为了回答你的问题，master 不会读取所有包含的数据，但它会在开始工作之前获取所有输入文件的状态。 Dataproc 将 属性 "mapreduce.input.fileinputformat.list-status.num-threads" 默认设置为 20 以帮助缩短此查找时间，但在 GCS 中仍会针对每个文件执行 RPC。
您似乎发现了这样一种情况，即使添加线程也无济于事，只会导致驱动程序更快地出现 OOM。
关于如何并行读取，我有两个想法。
但首先，有一点警告：这些解决方案都不是非常稳健的目录被包含在 glob 中。您可能希望防止目录出现在要读取的文件列表中。
第一个是使用 python 和 hadoop 命令行工具完成的（这也可以使用 gsutil 完成）。下面是一个示例，它可能看起来如何并在工作人员上执行文件列表，将文件内容成对读取并最终计算对（文件名，文件长度）：
from __future__ import print_function
from pyspark.rdd import RDD
from pyspark import SparkContext
import sys
import subprocess
def hadoop_ls(file_glob):
lines = subprocess.check_output(["/usr/bin/hadoop", "fs", "-ls", file_glob]).split("\n")
files = [line.split()[7] for line in lines if len(line) > 0]
return files
def hadoop_cat(file):
return subprocess.check_output(["/usr/bin/hadoop", "fs", "-cat", file]).decode("utf-8")
if __name__ == "__main__":
if len(sys.argv) < 2:
print("Provide a list of path globs to read.")
exit(-1)
sc = SparkContext()
# This is just for testing. You'll want to generate a list
# of prefix globs instead of having a list passed in from the
# command line.
globs = sys.argv[1:]
# Desired listing partition count
lpc = 100
# Desired 'cat' partition count, should be less than total number of files
cpc = 1000
files = sc.parallelize(globs).repartition(lpc).flatMap(hadoop_ls)
files_and_content = files.repartition(cpc).map(lambda f: [f, hadoop_cat(f)])
files_and_char_count = files_and_content.map(lambda p: [p[0], len(p[1])])
local = files_and_char_count.collect()
for pair in local:
print("File {} had {} chars".format(pair[0], pair[1]))
我会首先从这个子流程解决方案开始，然后尝试 hadoop_ls 和 hadoop_cat 调用的分区，看看您是否能得到可以接受的东西。
第二个解决方案更复杂，但可能会通过避免很多很多 exec 调用而产生性能更高的管道。
在第二个解决方案中，我们将编译一个特殊用途的帮助程序 jar，使用初始化操作将该 jar 复制到所有工作人员，最后使用驱动程序中的帮助程序。
我们的 scala jar 项目的最终目录结构将如下所示：
helper/src/main/scala/com/google/cloud/dataproc/support/PysparkHelper.scala
helper/build.sbt
在我们的 PysparkHelper.scala 文件中，我们将有一个小的 Scala class，其功能与我们上面的纯 python 解决方案一样多。首先，我们将创建一个文件 glob 的 RDD，然后是文件名的 RDD，最后是文件名和文件内容对的 RDD。
package com.google.cloud.dataproc.support
import collection.JavaConversions._
import org.apache.commons.io.IOUtils
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.api.java.{JavaPairRDD, JavaSparkContext}
import java.util.ArrayList
import java.nio.charset.StandardCharsets
class PysparkHelper extends Serializable {
def wholeTextFiles(
context: JavaSparkContext,
paths: ArrayList[String],
partitions: Int): JavaPairRDD[String, String] = {
val globRDD = context.sc.parallelize(paths).repartition(partitions)
// map globs to file names:
val filenameRDD = globRDD.flatMap(glob => {
val path = new Path(glob)
val fs: FileSystem = path.getFileSystem(new Configuration)
val statuses = fs.globStatus(path)
statuses.map(s => s.getPath.toString)
})
// Map file name to (name, content) pairs:
// TODO: Consider adding a second parititon count parameter to repartition before
// the below map.
val fileNameContentRDD = filenameRDD.map(f => {
Pair(f, readPath(f, new Configuration))
})
new JavaPairRDD(fileNameContentRDD)
}
def readPath(file: String, conf: Configuration) = {
val path = new Path(file)
val fs: FileSystem = path.getFileSystem(conf)
val stream = fs.open(path)
try {
IOUtils.toString(stream, StandardCharsets.UTF_8)
} finally {
stream.close()
}
}
}
helper/build.sbt 文件看起来像这样：
organization := "com.google.cloud.dataproc.support"
name := "pyspark_support"
version := "0.1"
scalaVersion := "2.10.5"
libraryDependencies += "org.apache.spark" % "spark-core_2.10" % "1.6.0" % "provided"
libraryDependencies += "org.apache.hadoop" % "hadoop-common" % "2.7.1" % "provided"
exportJars := true
然后我们可以使用 sbt 构建助手：
$ cd helper && sbt package
输出助手 jar 应该是 target/scala-2.10/pyspark_support_2.10-0.1.jar
我们现在需要将这个 jar 放到我们的集群上，为此，我们需要做两件事：1) 将 jar 上传到 GCS 和 2) 在 GCS 中创建一个初始化操作以将 jar 复制到集群节点.
为了便于说明，我们假设您的存储桶名为 MY_BUCKET（在此处插入适当的海象相关模因）。
$ gsutil cp target/scala-2.10/pyspark_support_2.10-0.1.jar gs://MY_BUCKET/pyspark_support.jar
创建一个初始化动作（我们称之为pyspark_init_action.sh，根据需要替换MY_BUCKET）：
#!/bin/bash
gsutil cp gs://MY_BUCKET/pyspark_support.jar /usr/lib/hadoop/lib/
最后将初始化动作上传到GCS：
$ gsutil cp pyspark_init_action.sh gs://MY_BUCKET/pyspark_init_action.sh
现在可以通过将以下标志传递给 gcloud 来启动集群：
--initialization-actions gs://MY_BUCKET/pyspark_init_action.sh
构建、上传和安装我们的新库后，我们可以最终从 pyspark 使用它：
from __future__ import print_function
from pyspark.rdd import RDD
from pyspark import SparkContext
from pyspark.serializers import PairDeserializer, UTF8Deserializer
import sys
class DataprocUtils(object):
@staticmethod
def wholeTextFiles(sc, glob_list, partitions):
"""
Read whole text file content from GCS.
:param sc: Spark context
:param glob_list: List of globs, each glob should be a prefix for part of the dataset.
:param partitions: number of partitions to use when creating the RDD
:return: RDD of filename, filecontent pairs.
"""
helper = sc._jvm.com.google.cloud.dataproc.support.PysparkHelper()
return RDD(helper.wholeTextFiles(sc._jsc, glob_list, partitions), sc,
PairDeserializer(UTF8Deserializer(), UTF8Deserializer()))
if __name__ == "__main__":
if len(sys.argv) < 2:
print("Provide a list of path globs to read.")
exit(-1)
sc = SparkContext()
globs = sys.argv[1:]
partitions = 10
files_and_content = DataprocUtils.wholeTextFiles(sc, globs, partitions)
files_and_char_count = files_and_content.map(lambda p: (p[0], len(p[1])))
local = files_and_char_count.collect()
for pair in local:
print("File {} had {} chars".format(pair[0], pair[1]))
谢谢！我尝试了第一种方法。它可以工作，但由于 exec 调用和 RPC/auth 开销，性能不是很好。在 32 节点集群上 运行 大约需要 10 个小时。我能够 运行 在 4 节点集群上使用带有 Amazon s3 连接器的 aws 上的数据块在 30 分钟内完成它。那里的开销似乎要少得多。我希望 Google 能提供一种更好的方法来将数据从 GCS 提取到 Spark。
我正在尝试使用 PySpark (Google Dataproc) 解析大约 100 万个 HTML 文件，并将相关字段写入压缩文件。每个 HTML 文件大约 200KB。因此，所有数据约为 200GB。
如果我使用数据的一个子集，下面的代码工作正常，但运行数小时然后在整个数据集上 运行 时崩溃。此外，未使用工作节点 (<5% CPU)，所以我知道存在一些问题。
我认为系统在从 GCS 摄取数据时出现问题。有一个更好的方法吗？另外，当我以这种方式使用 wholeTextFiles 时，主人是否会尝试下载所有文件然后将它们发送给执行者，还是让执行者下载它们？
def my_func(keyval):
keyval = (file_name, file_str)
return parser(file_str).__dict__
data = sc.wholeTextFiles("gs://data/*")
output = data.map(my_func)
output.saveAsTextFile("gs://results/a")
为了回答你的问题，master 不会读取所有包含的数据，但它会在开始工作之前获取所有输入文件的状态。 Dataproc 将 属性 "mapreduce.input.fileinputformat.list-status.num-threads" 默认设置为 20 以帮助缩短此查找时间，但在 GCS 中仍会针对每个文件执行 RPC。
您似乎发现了这样一种情况，即使添加线程也无济于事，只会导致驱动程序更快地出现 OOM。
关于如何并行读取，我有两个想法。
但首先，有一点警告：这些解决方案都不是非常稳健的目录被包含在 glob 中。您可能希望防止目录出现在要读取的文件列表中。
第一个是使用 python 和 hadoop 命令行工具完成的（这也可以使用 gsutil 完成）。下面是一个示例，它可能看起来如何并在工作人员上执行文件列表，将文件内容成对读取并最终计算对（文件名，文件长度）：
from __future__ import print_function
from pyspark.rdd import RDD
from pyspark import SparkContext
import sys
import subprocess
def hadoop_ls(file_glob):
lines = subprocess.check_output(["/usr/bin/hadoop", "fs", "-ls", file_glob]).split("\n")
files = [line.split()[7] for line in lines if len(line) > 0]
return files
def hadoop_cat(file):
return subprocess.check_output(["/usr/bin/hadoop", "fs", "-cat", file]).decode("utf-8")
if __name__ == "__main__":
if len(sys.argv) < 2:
print("Provide a list of path globs to read.")
exit(-1)
sc = SparkContext()
# This is just for testing. You'll want to generate a list
# of prefix globs instead of having a list passed in from the
# command line.
globs = sys.argv[1:]
# Desired listing partition count
lpc = 100
# Desired 'cat' partition count, should be less than total number of files
cpc = 1000
files = sc.parallelize(globs).repartition(lpc).flatMap(hadoop_ls)
files_and_content = files.repartition(cpc).map(lambda f: [f, hadoop_cat(f)])
files_and_char_count = files_and_content.map(lambda p: [p[0], len(p[1])])
local = files_and_char_count.collect()
for pair in local:
print("File {} had {} chars".format(pair[0], pair[1]))
我会首先从这个子流程解决方案开始，然后尝试 hadoop_ls 和 hadoop_cat 调用的分区，看看您是否能得到可以接受的东西。
第二个解决方案更复杂，但可能会通过避免很多很多 exec 调用而产生性能更高的管道。
在第二个解决方案中，我们将编译一个特殊用途的帮助程序 jar，使用初始化操作将该 jar 复制到所有工作人员，最后使用驱动程序中的帮助程序。
我们的 scala jar 项目的最终目录结构将如下所示：
helper/src/main/scala/com/google/cloud/dataproc/support/PysparkHelper.scala
helper/build.sbt
在我们的 PysparkHelper.scala 文件中，我们将有一个小的 Scala class，其功能与我们上面的纯 python 解决方案一样多。首先，我们将创建一个文件 glob 的 RDD，然后是文件名的 RDD，最后是文件名和文件内容对的 RDD。
package com.google.cloud.dataproc.support
import collection.JavaConversions._
import org.apache.commons.io.IOUtils
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.api.java.{JavaPairRDD, JavaSparkContext}
import java.util.ArrayList
import java.nio.charset.StandardCharsets
class PysparkHelper extends Serializable {
def wholeTextFiles(
context: JavaSparkContext,
paths: ArrayList[String],
partitions: Int): JavaPairRDD[String, String] = {
val globRDD = context.sc.parallelize(paths).repartition(partitions)
// map globs to file names:
val filenameRDD = globRDD.flatMap(glob => {
val path = new Path(glob)
val fs: FileSystem = path.getFileSystem(new Configuration)
val statuses = fs.globStatus(path)
statuses.map(s => s.getPath.toString)
})
// Map file name to (name, content) pairs:
// TODO: Consider adding a second parititon count parameter to repartition before
// the below map.
val fileNameContentRDD = filenameRDD.map(f => {
Pair(f, readPath(f, new Configuration))
})
new JavaPairRDD(fileNameContentRDD)
}
def readPath(file: String, conf: Configuration) = {
val path = new Path(file)
val fs: FileSystem = path.getFileSystem(conf)
val stream = fs.open(path)
try {
IOUtils.toString(stream, StandardCharsets.UTF_8)
} finally {
stream.close()
}
}
}
helper/build.sbt 文件看起来像这样：
organization := "com.google.cloud.dataproc.support"
name := "pyspark_support"
version := "0.1"
scalaVersion := "2.10.5"
libraryDependencies += "org.apache.spark" % "spark-core_2.10" % "1.6.0" % "provided"
libraryDependencies += "org.apache.hadoop" % "hadoop-common" % "2.7.1" % "provided"
exportJars := true
然后我们可以使用 sbt 构建助手：
$ cd helper && sbt package
输出助手 jar 应该是 target/scala-2.10/pyspark_support_2.10-0.1.jar
我们现在需要将这个 jar 放到我们的集群上，为此，我们需要做两件事：1) 将 jar 上传到 GCS 和 2) 在 GCS 中创建一个初始化操作以将 jar 复制到集群节点.
为了便于说明，我们假设您的存储桶名为 MY_BUCKET（在此处插入适当的海象相关模因）。
$ gsutil cp target/scala-2.10/pyspark_support_2.10-0.1.jar gs://MY_BUCKET/pyspark_support.jar
创建一个初始化动作（我们称之为pyspark_init_action.sh，根据需要替换MY_BUCKET）：
#!/bin/bash
gsutil cp gs://MY_BUCKET/pyspark_support.jar /usr/lib/hadoop/lib/
最后将初始化动作上传到GCS：
$ gsutil cp pyspark_init_action.sh gs://MY_BUCKET/pyspark_init_action.sh
现在可以通过将以下标志传递给 gcloud 来启动集群：
--initialization-actions gs://MY_BUCKET/pyspark_init_action.sh
构建、上传和安装我们的新库后，我们可以最终从 pyspark 使用它：
from __future__ import print_function
from pyspark.rdd import RDD
from pyspark import SparkContext
from pyspark.serializers import PairDeserializer, UTF8Deserializer
import sys
class DataprocUtils(object):
@staticmethod
def wholeTextFiles(sc, glob_list, partitions):
"""
Read whole text file content from GCS.
:param sc: Spark context
:param glob_list: List of globs, each glob should be a prefix for part of the dataset.
:param partitions: number of partitions to use when creating the RDD
:return: RDD of filename, filecontent pairs.
"""
helper = sc._jvm.com.google.cloud.dataproc.support.PysparkHelper()
return RDD(helper.wholeTextFiles(sc._jsc, glob_list, partitions), sc,
PairDeserializer(UTF8Deserializer(), UTF8Deserializer()))
if __name__ == "__main__":
if len(sys.argv) < 2:
print("Provide a list of path globs to read.")
exit(-1)
sc = SparkContext()
globs = sys.argv[1:]
partitions = 10
files_and_content = DataprocUtils.wholeTextFiles(sc, globs, partitions)
files_and_char_count = files_and_content.map(lambda p: (p[0], len(p[1])))
local = files_and_char_count.collect()
for pair in local:
print("File {} had {} chars".format(pair[0], pair[1]))
谢谢！我尝试了第一种方法。它可以工作，但由于 exec 调用和 RPC/auth 开销，性能不是很好。在 32 节点集群上 运行 大约需要 10 个小时。我能够 运行 在 4 节点集群上使用带有 Amazon s3 连接器的 aws 上的数据块在 30 分钟内完成它。那里的开销似乎要少得多。我希望 Google 能提供一种更好的方法来将数据从 GCS 提取到 Spark。