学习随笔 --SparkStreaming WordCount Java实现

package stuSpark.com;

import scala.Tuple2;


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 java.util.Arrays;



public class JavaSparkStreaming {
	 public static void main(String[] args){
	
	//JavaStreamingContext对象，它是流功能的主要入口点。使用两个执行线程创建一个本地StreamingContext，批处理间隔为1秒。
	//SparkStreaming 中local后必须为大于等于2的数字【即至少2条线程】。因为receiver 占了一个不断循环接收数据
	SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount");
	JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(1));
	
	// 创建一个DStream来表示来自TCP源的流数据，指定为主机名（例如localhost）和端口（例如9999）
	JavaReceiverInputDStream lines = jssc.socketTextStream("localhost", 9999);
	
	//linesDStream表示将从数据服务器接收的数据流。此流中的每条记录都是一行文本。然后用空格分割为单词
	// flatMap是一个DStream操作，通过从源DStream中的每个记录生成多个新记录来创建新的DStream。
	//使用FlatMapFunction对象定义了转换Java API中有许多这样的便利类可以帮助定义DStream转换。
	//DStream是RDD产生的模板，在Spark Streaming发生计算前，其实质是把每个Batch的DStream的操作翻译成为了RDD操作
	JavaDStream words = lines.flatMap(
			  new FlatMapFunction() {
			    @Override 
			    public Iterable call(String x) {
			      return (Iterable) Arrays.asList(x.split(" ")).iterator();
			    }
			  });
	//计算
	JavaPairDStream pairs = words.mapToPair(
	  new PairFunction() {
	    @Override 
	    public Tuple2 call(String s) {
	      return new Tuple2(s, 1);
	    }
	  });
	
	JavaPairDStream wordCounts = pairs.reduceByKey(
	  new Function2() {
	    @Override 
	    public Integer call(Integer i1, Integer i2) {
	      return i1 + i2;
	    }
	  });
	wordCounts.print();//将此DStream 中生成的每个RDD的前十个元素打印到控制台
	jssc.start();              // 开始
	jssc.awaitTermination();   // 等待计算终止
}
}