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

Question

例如，我们假设RDDresultrdd为：

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

当调用resultrdd.saveastextfile(S3n://path/to/soughere/)时，它将创建多个文件，如下所示：

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

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)

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

我想我可以连接到S3&在persist_to_s3函数本身中获取bucket，但是这个操作肯定会使AWS限制我的API使用，因为我有大量的RDD。

• 为了以我所描述的格式将数据快速持久化到S3中，我还能做什么？
• 如果使用repartition()/coalesce()是唯一正确的方法，那么哪一种更适合这种用法？

Answer 1

我想到的另一个选择是使用repartition（）/coalesce（）来减少分区的数量，然后用mapPartitions（）来预制上面的操作，这会起作用，但会慢得多。

Repartition&MapPartitions是相对较快的选项，但您提到过它很慢。我认为您可以考虑更传统的解决方案，如多线程读写器

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