spark streaming fileStream

Question

私はスパークストリーミングでプログラミングしていますが、スカラには問題があります。私は、この関数の定義は、このような関数のStreamingContext.fileStreamを使用する

をしようとしている：

def fileStream[K, V, F <: InputFormat[K, V]](directory: String)(implicit arg0: ClassManifest[K], arg1: ClassManifest[V], arg2: ClassManifest[F]): DStream[(K, V)] 

新しいファイル用のHadoop互換ファイルシステムを監視し、使用してそれらを読み込む入力ストリームを作成します。指定されたキー値タイプと入力フォーマット。ファイル名はで始まる。無視されます。 HDFSは、HDFSはHDFSは、新しいファイルの

を監視するために ディレクトリ HDFSディレクトリのファイル読み取るため F 入力形式のファイルを読み取るための V 値の種類を提出読み取るため K キーの種類私が合格する方法がわかりませんキーと値のタイプ。スパークストリーミングで マイコード：

val ssc = new StreamingContext(args(0), "StreamingReceiver", Seconds(1), 
    System.getenv("SPARK_HOME"), Seq("/home/mesos/StreamingReceiver.jar")) 

// Create a NetworkInputDStream on target ip:port and count the 
val lines = ssc.fileStream("/home/sequenceFile") 

JavaコードHadoopのファイルを書き込むために：

public class MyDriver { 

private static final String[] DATA = { "One, two, buckle my shoe", 
     "Three, four, shut the door", "Five, six, pick up sticks", 
     "Seven, eight, lay them straight", "Nine, ten, a big fat hen" }; 

public static void main(String[] args) throws IOException { 
    String uri = args[0]; 
    Configuration conf = new Configuration(); 
    FileSystem fs = FileSystem.get(URI.create(uri), conf); 
    Path path = new Path(uri); 
    IntWritable key = new IntWritable(); 
    Text value = new Text(); 
    SequenceFile.Writer writer = null; 
    try { 
     writer = SequenceFile.createWriter(fs, conf, path, key.getClass(), 
       value.getClass()); 
     for (int i = 0; i < 100; i++) { 
      key.set(100 - i); 
      value.set(DATA[i % DATA.length]); 
      System.out.printf("[%s]\t%s\t%s\n", writer.getLength(), key, 
        value); 
      writer.append(key, value); 
     } 
    } finally { 
     IOUtils.closeStream(writer); 
    } 
} 

}

Answer 1

をごfileStreamを使用したい場合は、あなたがすべて供給する必要があるとしていますそれを呼び出すときに3つのタイプのパラメータがあります。 Key、Value、InputFormatの種類を呼び出す前に知っておく必要があります。

val lines = ssc.fileStream[LongWritable, Text, TextInputFormat]("/home/sequenceFile") 

これら3種類は、あなたの種類のことが起こるならば、あなたはそれが任意の型を必要としないので、代わりにtextFileStreamを使用することがあります：あなたのタイプはLongWritable、TextとTextInputFormatた場合は、次のようなので、fileStreamを呼ぶだろう私が言及した3つのタイプを使用して、fileStreamに代理人を送ります。それを使用すると、次のようになります。あなたは

val lines = ssc.textFileStream("/home/sequenceFile") 

Answer 2

val filterF = new Function[Path, Boolean] { 
    def apply(x: Path): Boolean = { 
     val flag = if(x.toString.split("/").last.split("_").last.toLong < System.currentTimeMillis) true else false 
     return flag 
    } 
} 

val streamed_rdd = ssc.fileStream[LongWritable, Text, TextInputFormat]("/user/hdpprod/temp/spark_streaming_input",filterF,false).map(_._2.toString).map(u => u.split('\t'))