2

我是 Spark 的新手，我正在构建一个小型示例应用程序，它是一个 Spark 文件流式应用程序。我想要的只是一次读取整个文件，而不是逐行读取（我猜这就是 textFileStream 所做的）。

代码如下：

import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.StreamingContext._

import org.apache.hadoop.io.LongWritable
import org.apache.hadoop.io.Text
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat

import scalax.io._

object SampleXML{

    def main(args: Array[String]){

        val logFile = "/home/akhld/mobi/spark-streaming/logs/sample.xml"

        val ssc = new StreamingContext("spark://localhost:7077","XML Streaming Job",Seconds(5),"/home/akhld/mobi/spark-streaming/spark-0.8.0-incubating",List("target/scala-2.9.3/simple-project_2.9.3-1.0.jar"))

        val lines = ssc.fileStream[LongWritable, Text, TextInputFormat]("/home/akhld/mobi/spark-streaming/logs/")

        lines.print()

        lines.foreachRDD(rdd => {
          rdd.count()  // prints counts

        })


        ssc.start()


    }
}

此代码失败，异常表示：

[error] /home/akhld/mobi/spark-streaming/samples/samplexml/src/main/scala/SampleXML.scala:31: value foreachRDD is not a member of org.apache.spark.streaming.DStream[(org.apache.hadoop.io.LongWritable, org.apache.hadoop.io.Text)]
[error]         ssc.fileStream[LongWritable, Text, TextInputFormat]("/home/akhld/mobi/spark-streaming/logs/").foreachRDD(rdd => {
[error]                                                                                                       ^
[error] one error found
[error] (compile:compile) Compilation failed
[error] Total time: 3 s, completed Feb 3, 2014 7:32:57 PM

如果这不是在流中显示文件内容的正确方式，请帮我举个例子。我搜索了很多，但找不到合适的文件流。

4

2 回答 2

1

好吧，Spark Streaming 中的 textFileStream 更多是为了持续读取和处理正在写入目录中的文件。所以如果要一次性处理一个文件整体，直接用Spark会更简单！

 val lines = sparkContext.textFile(<file URL>)
 lines.foreach(line => println(line))

这将打印文件中的所有行。

于 2014-04-29T21:25:58.190 回答
0

另外，我相信这里的问题是您不能在 forEach 流内的 RDD 上调用 count() 。原因是，如果你这样做，我认为它会阻止 forEach 块的进度 - 并且流消费者停止工作。

我为此https://issues.apache.org/jira/browse/SPARK-4040创建了一个 JIRA 。

我认为当你在 forEach 块中引用 RDD 时，你可以对它们进行一些敏感的 API 调用，但还没有完全弄清楚所有细节。

于 2014-10-25T19:45:33.990 回答