0083.使用Scala和Java实战Spark Streaming开发第二讲Spark Streaming

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;
import scala.collection.parallel.ParIterableLike;

import java.util.Arrays;

/**
* Created by root on 16-4-19.
*/
public class WordCountOnline {
public static void main(String[] args){
/**第一步创建SparkConf
* 1.至少有两个线程，一个线程用于接收数据，一个线程用于处理数据
* 否则内存和磁盘都会无法承受
* 2.对于集群而言每个Executor,肯定不只有一个线程，对于处理Spark Streaming
* 的应用程序而言，一般分配几个Core比较合适，5个左右的Core比较合适
*/
SparkConf conf = new SparkConf().setMaster("local").setAppName("WordCountOnline");
JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(5));
/**
* 第三步创建Spark Streaming输入来源， input Stream
* 1.数据输入来源可以是：File,HDFS,Flume,Kafka,Socket等
* 2.该端口服务首先必须存在，不断产生数据
* 3.在生产环境中首先要判断是否有数据，如果没有数据，就不进行提交Job
*/
// JavaReceiverInputDStream lines = jsc.socketTextStream("local", 9999);
JavaDStream lines = jsc.textFileStream("/tmp/spark/");
/**
* 接下来就像基于RDD一样进行处理，基于DStream进行处理，DStream是RDD产生的模板或者说类，
* 在Spark Streaming发生计算前，其实际是把每个DStream翻译成RDD的操作
*/
JavaDStream words = lines.flatMap(new FlatMapFunction<String, String>() {
@Override
public Iterable call(String line) throws Exception {
return Arrays.asList(line.split(" "));
}
});

JavaPairDStream<String,Integer> pairs = words.mapToPair(new PairFunction<String, String, Integer>() {
@Override
public Tuple2 call(String word) throws Exception {
return new Tuple2<String, Integer>(word, 1);
}
});

JavaPairDStream<String,Integer> wordCounts = pairs.reduceByKey(new Function2<Integer, Integer, Integer>() {
@Override
public Integer call(Integer integer, Integer integer2) throws Exception {
return integer + integer2;
}
});
wordCounts.print();
jsc.start();
jsc.awaitTermination();

}
}

1.启动服务
2.进入目录：/tmp/spark/
3.创建文件：echo "A B B C" >> test1.txt
4.出现结果
-------------------------------------------
Time: 1461059835000 ms
-------------------------------------------
(B,2)
(A,1)
(C,1)