将每个 Spark RDD 条目分别保存到 S3

Question

我有一个（显然）加载和转换数据的 PySpark 应用程序。

我想将结果 RDD 保存到 S3，但 Spark 提供的 .saveAsTextFile() 函数不能满足我的要求，因为它将多个 RDD 条目写入一个文件。

例如，假设 RDD resultRDD 是：

[('cats', cats_data), ('dogs', dogs_data), ('flamingos', flamingos_data)]

调用resultRDD.saveAsTextFile(s3n://path/to/somewhere/)时，会创建多个文件，如下所示：

1. 000_part00 [containing cats_data & dogs_data]
2. 000_part01 [containing only flamingos_data]

请注意，创建的文件数量与 RDD 中的元素数量无关。此外，在打开文件之前，我什至不知道每个文件里面有什么。

相反，我想要创建的是以下输出：

1. cats [containing only cats_data]
2. dogs [containing only dogs_data]
2. flamingos [containing only flamingos_data]

我想我可能会使用 boto 的 S3Connection 并手动写入 S3，如下所示：

s3_connection = <connecting to S3 here>
bucket = s3_connection.get_bucket('animals_data')

def persist_to_s3(data_tuple):
   s3_key = bucket.create_key(key=data_tuple[0], bucket=bucket)
   s3_key.set_contents_from_string(data_tuple[1])

resultRDD.map(persist_to_s3)

不幸的是，连接和存储桶对象既不是可序列化的也不是线程安全的（我想），所以我不能像上面那样在节点之间共享连接。

我想我可能会连接到 S3 并在 persist_to_s3 函数本身中获取存储桶，但这个操作肯定会让 AWS 限制我的 API 使用，因为我有大量的 RDD。

澄清：我的实际数据集很大，而且键都是唯一的。因此，在这种情况下，按键重新分区将无济于事。

我想到的另一个选项是使用repartition() / coalesce() 以减少分区数量，然后使用mapPartitions() 执行上述操作，这将起作用但会慢得多。

• 我还能做些什么来快速以我描述的格式将数据保存到 S3？
• 如果使用repartition()/coalesce()是唯一正确的方法，哪种更适合这种用法？

Answer 1

我想到的另一个选项是使用 repartition() / coalesce() 以减少分区的数量，然后 使用 mapPartitions() 执行上述操作，这将起作用，但 会慢很多。

repartition&mapPartitions 是相对较快的选项，但您提到它很慢。我认为您可以考虑更传统的解决方案，例如multiple-thread reader/writer

1、使用你描述的格式写入结果数据；
2、使用多线程读写器模型将数据并行写入S3存储。类似“并行读取器--读取到-->并发阻塞队列---->并行写入器--写入----->S3”的工作流程。