apache-spark - 如何将转换后的数据从分区发送到 S3？

Question

我有一个RDD，它太大了collect。我对 RDD 应用了一系列转换，并希望将其转换后的数据直接从我的从节点上的分区发送到 S3。我目前操作如下:

val rdd:RDD = initializeRDD
val rdd2 = rdd.transform
rdd2.first // in order to force calculation of RDD
rdd2.foreachPartition sendDataToS3

不幸的是，发送到 S3 的数据是未转换的。 RDD 看起来和 initializeRDD 阶段完全一样。

这里是 sendDataToS3 的主体:

implicit class WriteableRDD[T](rdd:RDD[T]){

def transform:RDD[String] = rdd map {_.toString}

....
def sendPartitionsToS3(prefix:String) = {
  rdd.foreachPartition { p =>
    val filename = prefix+new scala.util.Random().nextInt(1000000)
    val pw = new PrintWriter(new File(filename))
    p foreach pw.println
    pw.close
    s3.putObject(S3_BUCKET, filename, new File(filename))
  }
  this
}

}

这是用 rdd.transform.sendPartitionsToS3(prefix) 调用的。

如何确保在 sendDataToS3 中发送的数据是转换后的数据？

Answer 1

我猜你的代码中有一个问题没有包含在问题中。

无论如何我都会回答，只是为了确保您了解 RDD.saveAsTextFile。您可以为其提供 S3 上的路径 (s3n://bucket/directory)，它会直接从执行程序将每个分区写入该路径。

我很难想象您什么时候需要实现自己的 sendPartitionsToS3 而不是使用 saveAsTextFile。