Spark：如何有效地将数据帧写入 S3

Question

我正在尝试找出使用(Py)Spark 将数据写入S3 的最佳方式。

我从 S3 存储桶读取似乎没有问题，但是当我需要写入时，它真的很慢。

我已经像这样启动了 spark shell（包括 hadoop-aws 包）：

AWS_ACCESS_KEY_ID=<key_id> AWS_SECRET_ACCESS_KEY=<secret_key> pyspark --packages org.apache.hadoop:hadoop-aws:3.2.0

这是示例应用程序

# Load several csv files from S3 to a Dataframe (no problems here)
df = spark.read.csv(path='s3a://mybucket/data/*.csv', sep=',')
df.show()

# Some processing
result_df = do_some_processing(df)
result_df.cache()
result_df.show()

# Write to S3
result_df.write.partitionBy('my_column').csv(path='s3a://mybucket/output', sep=',')  # This is really slow

当我尝试写入 S3 时，我收到以下警告：

20/10/28 15:34:02 WARN AbstractS3ACommitterFactory: Using standard FileOutputCommitter to commit work. This is slow and potentially unsafe.

我应该更改任何设置以高效写入 S3 吗？因为现在真的很慢，写100个小文件到S3大概需要10分钟。

Answer 1

事实证明，您必须手动指定提交者（否则将使用默认提交者，该提交者未针对 S3 进行优化）：

result_df \
    .write \
    .partitionBy('my_column') \
    .option('fs.s3a.committer.name', 'partitioned') \
    .option('fs.s3a.committer.staging.conflict-mode', 'replace') \
    .option("fs.s3a.fast.upload.buffer", "bytebuffer") \ # Buffer in memory instead of disk, potentially faster but more memory intensive
    .mode('overwrite') \
    .csv(path='s3a://mybucket/output', sep=',')

相关文档可以在这里找到：

• hadoop-aws
• hadoop-aws-committers