在spark流上下文中将RDD写入HDFS

Question

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

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，但使用此先决条件，我终于有了名为part-00000等的空输出文件...

使用arr.foreach（println）我能够看到传输的正确结果。

我的怀疑是火花在每个批次中尝试将数据写入相同的文件，删除之前写入的内容。我试图保存在一个动态命名文件夹中，如myRDD. saveAsTextFile（"文件夹"System.currentTimeMillis（）. toString（）），但总是只创建一个文件夹，输出文件仍然为空。

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

Answer 1

@vzamboni:Spark 1.5 dataframes api具有以下功能：

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

Answer 2

您使用的Spark Streaming的方式不是设计的。我建议在您的用例中放弃使用Spark，或者修改您的代码，使其以Spark的方式工作。将阵列收集到驱动程序违背了使用分布式引擎的目的，并使您的应用程序有效地成为一台机器（两台机器也会导致比仅在一台机器上处理数据更大的开销）。

您可以用数组做的一切，都可以用Spark来做。因此，只需在流中运行您的计算，分布在辅助角色上，然后使用DStream.saveAsTextFiles（）编写输出。您可以使用foreachRDDsaveAsParque（path， overwrite=true）写入单个文件。