在 spark-streaming 上下文中将 RDD 写入 HDFS答案

【问题标题】：write an RDD into HDFS in a spark-streaming context在 spark-streaming 上下文中将 RDD 写入 HDFS
【发布时间】：2015-09-19 21:40:35
【问题描述】：

我有一个带有 spark 1.2.0 的 spark 流环境，我从本地文件夹中检索数据，每次找到添加到文件夹中的新文件时，我都会执行一些转换。

val ssc = new StreamingContext(sc, Seconds(10))
val data = ssc.textFileStream(directory)

为了对 DStream 数据进行分析，我必须将其转换为数组

var arr = new ArrayBuffer[String]();
   data.foreachRDD {
   arr ++= _.collect()
}

然后我使用获得的数据来提取我想要的信息并将它们保存在 HDFS 上。

val myRDD  = sc.parallelize(arr)
myRDD.saveAsTextFile("hdfs directory....")

由于我确实需要使用 Array 操作数据，因此无法使用 DStream.saveAsTextFiles("...") 将数据保存在 HDFS 上（这样可以正常工作），我必须保存 RDD，但有了这个 preocedure，我终于有了名为 part- 的空输出文件00000 等...

使用arr.foreach(println) 我可以看到转换的正确结果。

我的怀疑是，spark 会尝试在每个批次中将数据写入相同的文件中，从而删除之前写入的内容。我试图保存在一个动态命名的文件夹中，比如myRDD.saveAsTextFile("folder" + System.currentTimeMillis().toString())，但总是只创建一个文件夹，并且输出文件仍然是空的。

如何在 spark-streaming 上下文中将 RDD 写入 HDFS？

【问题讨论】：

我想问题是您的 arr 并非对所有工作人员都可用。您是否尝试广播您的 arr，然后最终将其写入 hdfs？
因为我需要监控一个文件夹并拦截所有上传的新文件，并且火花流听起来是一个不错的解决方案。它不是单台机器，而是 2 台机器集群。现在我只是将文件作为文本编写，但将来我将不得不编写 parquet 文件，而且使用 Spark 非常简单
你会试试这个吗？ var arr = new ArrayBuffer[String](); val broadcasted = sc.broadcast(arr) data.foreachRDD { broadcasted ++= _.collect() } val myRDD = sc.parallelize(broadcasted ) myRDD.saveAsTextFile("hdfs directory....")
谢谢，我试过但没有结果。我想我必须只使用 DStream

标签： scala hadoop apache-spark hdfs spark-streaming

【解决方案1】：

您使用 Spark Streaming 的方式并非设计。我会建议您将 Spark 用于您的用例，或者调整您的代码，使其以 Spark 方式工作。将数组收集到驱动程序会破坏使用分布式引擎的目的，并使您的应用程序有效地单机（两台机器也会导致比仅在单台机器上处理数据更多的开销）。

您可以对数组执行的所有操作，都可以使用 Spark。因此，只需在流中运行您的计算，分布在工作人员上，然后使用DStream.saveAsTextFiles() 编写您的输出。您可以使用foreachRDD + saveAsParquet(path, overwrite = true) 写入单个文件。

【讨论】：

谢谢，我完全明白你的意思，我会尝试改变 trasform 逻辑以使用 DStream。您知道是否可以在每批中使用火花流将记录保存在同一个文件中？现在我得到一个新文件夹，每个批处理间隔都有新文件。
是的，使用 foreachRDD + saveAsParquet 可以选择覆盖。

【解决方案2】：

@vzamboni：Spark 1.5+ dataframes api 有这个功能：

dataframe.write().mode(SaveMode.Append).format(FILE_FORMAT).partitionBy("parameter1", "parameter2").save(path);

【讨论】：