Spark 'saveAsTextFile' 到 S3：无法控制带有 'coalesce' 的文件数量

Posted 2023-04-15

【中文标题】Spark \'saveAsTextFile\' 到 S3：无法控制带有 \'coalesce\' 的文件数量

【英文标题】：Spark 'saveAsTextFile' to S3: Can't control number of files with 'coalesce'Spark 'saveAsTextFile' 到 S3：无法控制带有 'coalesce' 的文件数量

【发布时间】：2017-01-20 20:18:25

【问题描述】：

将 Python 3 与 PySpark 和 Spark 1.6.0 结合使用。我读过saveAsTextFile() 创建的文件数等于RDD 分区数。但是，我专门将 RDD 合并到 16 个分区，但只有 1 个文件被写入 S3 (part-00000.gz)。我做错了什么？

这是我正在使用的代码：

conf = SparkConf()
sc = SparkContext(conf=conf)
sc.setLogLevel('WARN')
sc._jsc.hadoopConfiguration().set('fs.s3a.access.key', AWS_ACCESS_KEY)
sc._jsc.hadoopConfiguration().set('fs.s3a.secret.key', AWS_SECRET_KEY)
sqlContext = HiveContext(sc)

tbl = sqlContext.table(TABLE)
tbl.limit(1000000).toJSON().coalesce(16).saveAsTextFile(S3A_BUCKET_URL, compressionCodecClass="org.apache.hadoop.io.compress.GzipCodec")

原来的TABLE 是 Parquet 存储在大约 11,000 个文件中（我假设这等于 Spark 分区？）。当我不在整个表上使用 limit() 和 coalesce() 时，它确实会尝试在 S3 上创建数千个小文件，这需要很长时间，而且当我希望使用更少的大文件时就没有必要了。

【参考方案1】：

这是因为您使用了limit。至于现在（有an ongoing discussion on the developers list，所以将来可能会改变）limit 将所有数据重新分区到单个分区。由于coalesce 只能减少分区数量，因此没有任何效果。

出于性能原因，最好使用sample 和coalesce。例如：

from operator import truediv

df.cache()
n = ... # Number of records to take
m = df.count()

df.sample(withReplacement=False, fraction=truediv(n / m))

但如果您想使用精确的limit，则必须使用repartition 而不是coalesce。

【讨论】：

可以理解limit 是罪魁祸首，而我忘记了sample！谢谢。