如何使用火花流读取文件并使用 Scala 写入简单文件？答案

【问题标题】：How to read a file using sparkstreaming and write to a simple file using Scala?如何使用火花流读取文件并使用 Scala 写入简单文件？
【发布时间】：2017-04-28 09:28:56
【问题描述】：

我正在尝试使用 scala SparkStreaming 程序读取文件。该文件存储在我的本地计算机上的一个目录中，并尝试将其作为新文件写入我的本地计算机本身。但是，每当我编写流并将其存储为镶木地板时，我最终都会得到空白文件夹。

这是我的代码：

 Logger.getLogger("org").setLevel(Level.ERROR)
 val spark = SparkSession
             .builder()
             .master("local[*]")
             .appName("StreamAFile")
             .config("spark.sql.warehouse.dir", "file:///C:/temp")
             .getOrCreate()
 
         
 import spark.implicits._            
 val schemaforfile = new StructType().add("SrNo",IntegerType).add("Name",StringType).add("Age",IntegerType).add("Friends",IntegerType)
             
 val file = spark.readStream.schema(schemaforfile).csv("C:\\SparkScala\\fakefriends.csv")  

 file.writeStream.format("parquet").start("C:\\Users\\roswal01\\Desktop\\streamed") 
 
 spark.stop()

我的代码中是否有任何遗漏或我出错的地方？

我也尝试从 hdfs 位置读取此文件，但相同的代码最终没有在我的 hdfs 上创建任何输出文件夹。

【问题讨论】：

标签： scala apache-spark spark-streaming parquet

【解决方案1】：

你错了：

val file = spark.readStream.schema(schemaforfile).csv("C:\\SparkScala\\fakefriends.csv")

csv() 函数应该有目录路径作为参数。它将扫描此目录并在将所有新文件移动到此目录时读取它们

对于检查点，您应该添加

.option("checkpointLocation", "path/to/HDFS/dir")

例如：

val query = file.writeStream.format("parquet")
    .option("checkpointLocation", "path/to/HDFS/dir")
    .start("C:\\Users\\roswal01\\Desktop\\streamed") 

query.awaitTermination()

【讨论】：

file.writeStream.parquet("C:\\Users\\roswal01\\Desktop\\streamed").start() 这里 parquet 似乎给出了错误，它表明它不是 DataStreamWriter 的成员。我错过了导入任何包吗？
它也给出了一个检查点错误。 checkpointLocation must be specified either through option("checkpointLocation", ...) or SparkSession.conf.set("spark.sql.streaming.checkpointLocation", ...);
@RohanOswal 文档似乎已过时 - 抱歉！我已经从答案中删除了这个 parquet() 函数。我还添加了如何进行检查点的信息，它曾经 - 并且可能仍然 - 并不总是需要，这就是我没有包含它的原因
我的输出仍然是空白文件夹。该文件夹已创建，但其中没有文件，只有一个 _spark_metadata 文件夹，该文件夹再次为空白。
@RohanOswal 很奇怪。我会再次调查并发布更新:)