
205
10.5 入力ソース
10-27
Java
DStream
SequenceFile
ー
JavaPairDStream<Text, LongWritable> writableDStream = ipDStream.mapToPair(
new PairFunction<Tuple2<String, Long>, Text, LongWritable>() {
public Tuple2<Text, LongWritable> call(Tuple2<String, Long> e) {
return new Tuple2(new Text(e._1()), new LongWritable(e._2()));
}});
class OutFormat extends SequenceFileOutputFormat<Text, LongWritable> {};
writableDStream.saveAsHadoopFiles(
"outputDir", "txt", Text.class, LongWritable.class, OutFormat.class);
最後に、汎用の出力操作である
foreachRDD()
を使えば、任意の演算処理を
DStream
内の
RDD
群に対して実行できます。これは、それぞれの
RDD
に対して処理を行えるという点で、
transform()
に似ています。
foreachRDD()
では、
Spark
で使えるすべてのアクションを再利用でき ...