Pyspark 按键写入多个输出，无需分区答案

【问题标题】：Pyspark write multiple outputs by key without partitionPyspark 按键写入多个输出，无需分区
【发布时间】：2018-12-06 03:46:22
【问题描述】：

我有一个 PySpark 数据框，其中包含 600 万人的记录，每个人都有一个 userid。每个 userid 有 2000 个条目。我想将我的每个 userid 的数据保存到一个单独的 csv 文件中，并以 userid 作为名称。

我有一些代码可以做到这一点，taken from the solution to this question。但是，据我了解，代码将尝试对 600 万个 id 中的每一个进行分区。我实际上并不关心这一点，因为我要将这些文件中的每一个写入另一个非 HDFS 服务器。

我应该注意，该代码适用于少数 userids（最多 3000 个），但它在全部 600 万个时失败。

代码：

output_file = '/path/to/some/hdfs/location'
myDF.write.partitionBy('userid').mode('overwrite').format("csv").save(output_file)

当我运行上面的代码时，它需要 WEEKS 才能运行，其中大部分时间都花在了编写步骤上。我认为这是因为分区的数量。即使我手动将分区数指定为较小的值，它仍然需要很长时间才能执行。

问题：有没有办法将每个userids 数据保存到一个命名良好（文件名=userid）的文件中而不进行分区？

【问题讨论】：

这听起来不像是 hdfs 的好用例。你不能改变你的文件系统吗？
我不知道为单个任务更改文件系统是否是最佳解决方案。还有其他文件系统更擅长处理 2000x600 万行数据吗？
对不起，我写的时候真的很着急，我离开后才意识到这是多么愚蠢：D 也许你可以改变你的保存方式。你真的需要单独的文件吗？你真的需要文件吗？也许某种数据库会更好？
我的计划是对 2000 个文档中的每一个（针对所有 600 万用户）进行单独分析。查询这种大小的数据库需要做很多工作。我已经尝试上传到 MySQL，但我的服务器无法处理创建可以加快任何查询的正确索引。如果我处理一次数据并将 2000 个文档中的每一个都放在一个单独的文件中，那么我不需要任何特殊的选择或排序。

标签： python apache-spark pyspark

【解决方案1】：

考虑到要求，确实有很大的改进希望。 HDFS 不是为处理非常小的文件而设计的，如果您尝试同时打开 600 万个文件描述符，几乎任何文件系统都会受到挑战。

如果你还没有在写之前调用 repartition，你可以稍微改进一下：

(myDF
    .repartition('userid')
    .write.partitionBy('userid').mode('overwrite').format("csv").save(output_file))

如果每个文件可以接受多个 id，则可以使用持久表和分桶

myDFA
  .write
  .bucketBy(1024, 'userid')  # Adjust numBuckets if needed
  .sortBy('userid')
  .mode('overwrite').format("csv")
  .saveAsTable(output_table))

分别处理每个文件，获取连续的数据块。

最后，如果纯文本输出不是硬性要求，您可以使用userid 的任何分片数据库和分区数据。

【讨论】：

我认为这是在正确的轨道上。但是，我会以镶木地板格式存储数据，并按此处所示进行排序。这样，每个用户 ID 的分析查询应该会更容易。也许你可以完全移除桶。由于洗牌，对数据进行排序会很慢 - 但最终你会错过这种排序。
第一个建议是对我的解决方案略有改进，但仍然不是我需要的。第二个建议说“每个文件有多个 id”，这与我的要求完全相反。（并不是说这是个混蛋，我很感激帮助。）在发布此答案之前，我在 cmets 中针对原始问题提出的第三个建议：是的，我可以上传到数据库（例如 MySQL），但将索引放在 2000 x 600 万条记录超出了我的系统的处理能力。