json - 使用 spark-submit 从 google dataproc spark cluster 读取 GCP 中的 JSON(zipped .gz) 时，未使用所有执行程序

Question

我刚刚使用 GCP(dataproc) 和 pyspark 了解了这个奇妙的大数据和云技术世界。我有 ~5 GB 尺码 JSON 包含 的文件(压缩，gz 文件) ~500 万 记录，我需要读取每一行并只处理满足特定条件的那些行。我有我的工作代码，我用 --num-partitions=5 发出了一个 spark-submit 但仍然 只有一名 worker 用于执行 Action 。

这是我正在使用的 spark-submit 命令:

spark-submit --num-executors 5 --py-files /home/user/code/dist/package-0.1-py3.6.egg job.py

工作.py:

path = "gs://dataproc-bucket/json-files/data_5M.json.gz"
mi = spark.read.json(path)
inf_rel = mi.select(mi.client_id,
                    mi.user_id,
                    mi.first_date,
                    F.hour(mi.first_date).alias('hour'),
                    mi.notes).rdd.map(foo).filter(lambda x: x)
inf_relevance = inf_rel.map(lambda l: Row(**dict(l))).toDF()
save_path = "gs://dataproc-bucket/json-files/output_5M.json"
inf_relevance.write.mode('append').json(save_path)
print("END!!")

数据处理配置:
(我现在正在使用免费帐户，一旦我获得工作解决方案将添加更多内核和执行程序)

(Debian 9、Hadoop 2.9、Spark 2.4)
主节点:2 个 vCPU，7.50 GB 内存
主磁盘大小:32 GB
5 个工作节点:1 个 vCPU，3.75 GB 内存
主磁盘类型:32 GB

在 spark-submit 之后，我可以在 Web UI 中看到添加了 5 个执行程序，但只有 1 个执行程序保持事件状态并执行所有任务，其余 4 个被释放。


我做了我的研究，大多数问题都是关于通过 JDBC 访问数据的。

请建议我在这里缺少什么。

附言最终我会读取 64 个 5 GB 的 json 文件，因此可能会使用 8 个核心 * 100 个 worker 。

Answer 1

最好的办法是预处理输入。给定一个输入文件，spark.read.json(...将创建一个任务来读取和解析 JSON 数据，因为 Spark 无法提前知道如何对其进行并行化。如果您的数据采用行分隔的 JSON 格式 ( http://jsonlines.org/ )，最好的做法是事先将其拆分为可管理的块:

path = "gs://dataproc-bucket/json-files/data_5M.json"
# read monolithic JSON as text to avoid parsing, repartition and *then* parse JSON
mi = spark.read.json(spark.read.text(path).repartition(1000).rdd)
inf_rel = mi.select(mi.client_id,
                   mi.user_id,
                   mi.first_date,
                   F.hour(mi.first_date).alias('hour'),
                   mi.notes).rdd.map(foo).filter(lambda x: x)
inf_relevance = inf_rel.map(lambda l: Row(**dict(l))).toDF()
save_path = "gs://dataproc-bucket/json-files/output_5M.json"
inf_relevance.write.mode('append').json(save_path)
print("END!!")

您在此处的初始步骤 ( spark.read.text(... ) 仍将作为单个任务出现瓶颈。如果您的数据不是行分隔的，或者(尤其是!)您预计需要多次处理这些数据，您应该想办法在让 Spark 参与之前将 5GB JSON 文件转换为 1000 个 5MB JSON 文件。