如何在 Google Datproc 中优化 Hadoop MapReduce 压缩 Spark 输出？答案

【问题标题】：How to optimize Hadoop MapReduce compressing Spark output in Google Datproc?如何在 Google Datproc 中优化 Hadoop MapReduce 压缩 Spark 输出？
【发布时间】：2018-12-27 13:35:36
【问题描述】：

目标：Cassandra 中的数百万行需要尽可能快速有效地提取并压缩到单个文件中（每天）。

当前设置使用 Google Dataproc 集群运行 Spark 作业，将数据直接提取到 Google Cloud Storage 存储桶中。我尝试了两种方法：

使用（现已弃用）FileUtil.copyMerge() 将大约 9000 个 Spark 分区文件组合成一个未压缩的文件，然后提交 Hadoop MapReduce 作业以压缩该单个文件。
保留大约 9000 个 Spark 分区文件作为原始输出，并提交 Hadoop MapReduce 作业以将这些文件合并并压缩为单个文件。

一些工作细节：大约 8 亿行。 Spark作业输出的大约9000个Spark分区文件。 Spark 作业在 1 个 Master、4 个 Worker（4 个 vCPU，每个 15GB）Dataproc 集群上完成运行大约需要一个小时。默认 Dataproc Hadoop 块大小，我认为是 128MB。

一些 Spark 配置细节：

spark.task.maxFailures=10
spark.executor.cores=4

spark.cassandra.input.consistency.level=LOCAL_ONE
spark.cassandra.input.reads_per_sec=100
spark.cassandra.input.fetch.size_in_rows=1000
spark.cassandra.input.split.size_in_mb=64

Hadoop 作业：

hadoop jar file://usr/lib/hadoop-mapreduce/hadoop-streaming-2.8.4.jar
-Dmapred.reduce.tasks=1
-Dmapred.output.compress=true
-Dmapred.compress.map.output=true
-Dstream.map.output.field.separator=,
-Dmapred.textoutputformat.separator=,
-Dmapred.output.compression.codec=org.apache.hadoop.io.compress.GzipCodec
-input gs://bucket/with/either/single/uncompressed/csv/or/many/spark/partition/file/csvs
-output gs://output/bucket
-mapper /bin/cat
-reducer /bin/cat
-inputformat org.apache.hadoop.mapred.TextInputFormat
-outputformat org.apache.hadoop.mapred.TextOutputFormat

Spark 作业大约需要 1 小时才能将 Cassandra 数据提取到 GCS 存储桶。使用 FileUtil.copyMerge() 增加了大约 45 分钟，由 Dataproc 集群执行，但资源未充分利用，因为它似乎使用 1 个节点。压缩该单个文件的 Hadoop 作业又花费了 50 分钟。这不是最佳方法，因为集群必须保持更长时间，即使它没有使用其全部资源。

该作业的信息输出：

INFO mapreduce.Job: Counters: 55
File System Counters
    FILE: Number of bytes read=5072098452
    FILE: Number of bytes written=7896333915
    FILE: Number of read operations=0
    FILE: Number of large read operations=0
    FILE: Number of write operations=0
    GS: Number of bytes read=47132294405
    GS: Number of bytes written=2641672054
    GS: Number of read operations=0
    GS: Number of large read operations=0
    GS: Number of write operations=0
    HDFS: Number of bytes read=57024
    HDFS: Number of bytes written=0
    HDFS: Number of read operations=352
    HDFS: Number of large read operations=0
    HDFS: Number of write operations=0
Job Counters 
    Killed map tasks=1
    Launched map tasks=353
    Launched reduce tasks=1
    Rack-local map tasks=353
    Total time spent by all maps in occupied slots (ms)=18495825
    Total time spent by all reduces in occupied slots (ms)=7412208
    Total time spent by all map tasks (ms)=6165275
    Total time spent by all reduce tasks (ms)=2470736
    Total vcore-milliseconds taken by all map tasks=6165275
    Total vcore-milliseconds taken by all reduce tasks=2470736
    Total megabyte-milliseconds taken by all map tasks=18939724800
    Total megabyte-milliseconds taken by all reduce tasks=7590100992
Map-Reduce Framework
    Map input records=775533855
    Map output records=775533855
    Map output bytes=47130856709
    Map output materialized bytes=2765069653
    Input split bytes=57024
    Combine input records=0
    Combine output records=0
    Reduce input groups=2539721
    Reduce shuffle bytes=2765069653
    Reduce input records=775533855
    Reduce output records=775533855
    Spilled Records=2204752220
    Shuffled Maps =352
    Failed Shuffles=0
    Merged Map outputs=352
    GC time elapsed (ms)=87201
    CPU time spent (ms)=7599340
    Physical memory (bytes) snapshot=204676702208
    Virtual memory (bytes) snapshot=1552881852416
    Total committed heap usage (bytes)=193017675776
Shuffle Errors
    BAD_ID=0
    CONNECTION=0
    IO_ERROR=0
    WRONG_LENGTH=0
    WRONG_MAP=0
    WRONG_REDUCE=0
File Input Format Counters 
    Bytes Read=47132294405
File Output Format Counters 
    Bytes Written=2641672054

我预计这种方法的性能与其他方法一样好或更好，但它的性能要差得多。 Spark 作业保持不变。跳过 FileUtil.copyMerge() 并直接跳入 Hadoop MapReduce 作业……一个半小时后，作业的地图部分只有大约 50%。那时工作被取消了，因为很明显它不可行。

我可以完全控制 Spark 作业和 Hadoop 作业。我知道我们可以创建一个更大的集群，但我宁愿在确保作业本身得到优化之后才这样做。任何帮助表示赞赏。谢谢。

【问题讨论】：

标签： apache-spark hadoop google-cloud-dataproc

【解决方案1】：

您能否提供更多有关您的 Spark 作业的详细信息？您使用的是什么 Spark API - RDD 或 Dataframe？为什么不在 Spark 中完全执行合并阶段（使用 repartition().write()）并避免 Spark 和 MR 作业的链接？

【讨论】：