emr-5.20.0 Spark 2.4.0 将 parquet 写入 s3 需要太多时间

Question

我正在使用：

• emr-5.20.0
• Hadoop 2.8.5
• Spark 2.4.0

我的一项 Spark 作业将 parquet 数据写入 s3，这似乎在 20-30 分钟内完成，90% 的处理完成，但最后 5-10% 需要 2 小时才能完成。我阅读了很多论坛，并了解到 EMR 正在使用优化的输出提交程序，但它仍然需要太多时间。我正在尝试配置自定义提交者，但该作业始终使用 EmrOptimizedSparkSqlParquetOutputCommitter，我如何禁用此功能并添加自定义提交者......以下是日志：

19/01/12 23:17:11 INFO FileOutputCommitter: File Output Committer Algorithm version is 2
19/01/12 23:17:11 INFO FileOutputCommitter: FileOutputCommitter skip cleanup _temporary folders under output directory:false, ignore cleanup failures: true
19/01/12 23:17:11 INFO SQLHadoopMapReduceCommitProtocol: Using user defined output committer class com.amazon.emr.committer.EmrOptimizedSparkSqlParquetOutputCommitter
19/01/12 23:17:11 INFO EmrOptimizedParquetOutputCommitter: EMR Optimized Committer: ENABLED
19/01/12 23:17:11 INFO EmrOptimizedParquetOutputCommitter: Using output committer class org.apache.hadoop.mapreduce.lib.output.FileSystemOptimizedCommitter
19/01/12 23:17:11 INFO FileOutputCommitter: File Output Committer Algorithm version is 2
19/01/12 23:17:11 INFO FileOutputCommitter: FileOutputCommitter skip cleanup _temporary folders under output directory:false, ignore cleanup failures: true
19/01/12 23:17:11 INFO FileOutputCommitter: File Output Committer Algorithm version is 2
19/01/12 23:17:11 INFO FileOutputCommitter: FileOutputCommitter skip cleanup _temporary folders under output directory:false, ignore cleanup failures: true
19/01/12 23:17:11 INFO SQLHadoopMapReduceCommitProtocol: Using output committer class com.amazon.emr.committer.EmrOptimizedSparkSqlParquetOutputCommitter
19/01/12 23:17:11 INFO FileSystemOptimizedCommitter: Nothing to setup as successful task attempt outputs are written directly.```

How do i disbale EMR to not use its own optimized EmrOptimizedSparkSqlParquetOutputCommitter

--conf spark.hadoop.mapred.output.committer.class = ai.peritus.training.preprocess.PeritusS3PartitionedOutputFormat 
--conf spark.hadoop.mapreduce.use.directfileoutputcommitter=false 
--conf spark.hadoop.spark.sql.parquet.output.committer.class= com.netflix.bdp.s3.S3PartitionedOutputCommitter 
--conf mapreduce.fileoutputcommitter.algorithm.version=1

Answer 1

我来自 EMR 团队，所以我对这个功能有点偏见，尽管有一些证据表明它对其他客户也很有效。例如，请参阅https://stackoverflow.com/a/54350777/2205987。

但是，我将首先回答您提出的问题。为了使用您自己的提交者类，您需要设置spark.sql.parquet.output.committer.class 属性。在上面的示例中，您错误地使用了spark.hadoop.spark.sql.parquet.output.committer.class。

也就是说，我认为我们首先应该确认您的应用程序甚至首先使用了 EMRFS S3 优化的提交程序。如果您在作业结束时遇到缓慢，则可能是优化的提交程序实际上没有被使用，或者您的作业中存在其他一些缓慢的原因，例如数据倾斜导致任务运行时间很长。

在某些情况下，即使启用了 EMRFS S3 优化的提交程序，它也不会实际使用。这将有助于了解有关您的应用程序的更多信息，例如一些示例代码。此外，如果您能够提供示例集群 id (j-ABC123)，它将帮助 EMR 的人员调试您的问题。