在 Windows 系统中打印流（Spark 流）的内容答案

【问题标题】：Print the content of streams (Spark streaming) in Windows system在 Windows 系统中打印流（Spark 流）的内容
【发布时间】：2018-10-30 16:50:14
【问题描述】：

我只想将流的内容打印到控制台。我编写了以下代码，但它不打印任何内容。任何人都可以帮助我在 Spark 中将文本文件作为流读取？是否存在与 Windows 系统相关的问题？

public static void main(String[] args) throws Exception {

     SparkConf sparkConf = new SparkConf().setAppName("My app")
        .setMaster("local[2]")
        .setSparkHome("C:\\Spark\\spark-1.5.1-bin-hadoop2.6")
        .set("spark.executor.memory", "2g");

    JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, Durations.seconds(2));

    JavaDStream<String> dataStream = jssc.textFileStream("C://testStream//copy.csv");
    dataStream.print();

    jssc.start();
    jssc.awaitTermination();
}

更新：copy.csv 的内容是

0,0,12,5,0
0,0,12,5,0
0,1,2,0,42
0,0,0,0,264
0,0,12,5,0

【问题讨论】：

copy.csv的内容是什么？
它是逗号分隔的文件。我用内容更新了我的问题

标签： java apache-spark spark-streaming

【解决方案1】：

textFileStream 用于监控 hadoop 兼容目录。此操作将监视提供的目录，并且当您在提供的目录中添加新文件时，它将从新添加的文件中读取/流式传输数据。

您无法使用textFileStream 读取文本/ csv 文件，或者我会说您不需要流式传输，以防您只是读取文件。

我的建议是监控某个目录（可能是 HDFS 或本地文件系统），然后使用textFileStream 添加文件并捕获这些新文件的内容。

可能在您的代码中，您可以将"C://testStream//copy.csv" 替换为C://testStream"，一旦您的Spark Streaming 作业启动并运行，然后将文件copy.csv 添加到C://testStream 文件夹并在Spark 控制台上查看输出。

或

也许您可以编写另一个命令行 Scala/Java 程序来读取文件并将内容扔到 Socket（在某个 PORT# 处），然后您可以利用 socketTextStream 来捕获和读取数据。读取数据后，您可以进一步应用其他转换或输出操作。

你也可以考虑利用Flume

更多详情请参考API Documentation

【讨论】：

实际上，我不仅在阅读文件，我还编写了一小部分代码，只是为了让我的问题显而易见。实际上，我想将流的内容发送到朴素贝叶斯模型进行预测（机器学习），但不幸的是，我不能将文件作为流读取。您能否向我解释一下如何在 Spark 中将 csv 文件作为流处理？我不明白如何在 Windows 中监控 HDFS 或本地文件系统？我试图用 C://testStream 替换“C://testStream//copy.csv”，但是当我将文件添加到目录时这不起作用..!!!
我已经详细说明了阅读文本文件的答案，就textFileStream 而言，请确保您正在创建具有唯一/不同名称的新文件，并且内容也应该是新的。不要重命名或复制具有相同内容的文件。那是行不通的。

【解决方案2】：

这在 Windows 7 和 Spark 1.6.3 上对我有用：（删除其余代码，重要的是如何定义要监视的文件夹）

val ssc = ...
val lines = ssc.textFileStream("file:///D:/tmp/data")
...
print

...

这监控目录 D:/tmp/data，ssc 是我的流上下文

步骤：

在 D:/tmp/data 中创建一个文件 1.txt
输入一些文字
启动 spart 应用程序
将文件重命名为 data.txt（我相信任何任意名称都可以，只要在 spark 监视目录时更改它即可）

我注意到的另一件事是我必须将行分隔符更改为 Unix 样式（使用 Notepad++），否则文件不会被拾取。

【讨论】：

【解决方案3】：

试试下面的代码，它可以工作：

JavaDStream<String> dataStream = jssc.textFileStream("file:///C:/testStream/");

【讨论】：