2

私はSparkの初心者で、Spark fileStreamingの小さなサンプルアプリケーションを構築しています。私が望んでいたのは、行ごとに読み取るのではなく、ファイル全体を一度に読み取ることだけです(これが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

これがストリーム内のファイルの内容を表示する正しい方法でない場合は、例を教えてください。私はたくさん検索しましたが、fileStream を使用する適切なものを見つけることができませんでした。

4

2 に答える 2

1

さて、Spark Streaming の textFileStream は、ディレクトリ内に書き込まれているファイルを継続的に読み取り、処理することを目的としています。したがって、1 つのファイル全体を一度に処理する必要がある場合は、Spark を直接使用する方が簡単です!

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

これにより、ファイル内のすべての行が出力されます。

于 2014-04-29T21:25:58.190 に答える