Window函式，可以統計最近一段時間的資料，使用Window函式載入成DStream：DStream.window("視窗長度","滑動間隔")reduceByKeyAndWindow視窗長度：必須是BathInterval的整數倍滑動間隔：必須是BatchInterval的整數倍/** * 1、local的模擬執行緒數必須大於等於2 因為一條執行緒被receiver(接受資料的執行緒)佔用，另外一個執行緒是job執行 * 2、Durations時間的設定，就是我們能接受的延遲度，這個我們需要根據叢集的資源情況以及監控每一個job的執行時間來調節出最佳時間。 * 3、 建立JavaStreamingContext有兩種方式 （sparkconf、sparkcontext） * 4、業務邏輯完成後，需要有一個output operator * 5、JavaStreamingContext.start()straming框架啟動之後是不能在次新增業務邏輯 * 6、JavaStreamingContext.stop()無參的stop方法會將sparkContext一同關閉，stop(false) ,預設為true，會一同關閉 * 7、JavaStreamingContext.stop()停止之後是不能在呼叫start */public class WindowOperator { @SuppressWarnings("resource") public static void main(String[] args) { SparkConf conf = new SparkConf(); //local[2]模擬執行緒數 conf.setMaster("local[2]").setAppName("ww"); //在此處設定最小間隔時間durations JavaStreamingContext jsc = new JavaStreamingContext(conf,Durations.seconds(5)); jsc.sparkContext().setLogLevel("WARN"); jsc.checkpoint("./check"); //設定監聽的節點和ip埠 JavaReceiverInputDStream<String> sts = jsc.socketTextStream("CentOS16",9999);//設定檔案保留時間，之後得而RDD使用Window時，只會取到最近60秒的資料 JavaDStream<String> window = sts.window(Durations.seconds(60),Durations.seconds(5)); JavaDStream<String> flatMap = sts.flatMap(new FlatMapFunction<String, String>() { private static final long serialVersionUID = 1L; @Override public Iterable<String> call(String arg0) throws Exception { return Arrays.asList(arg0.split(" ")); } }); JavaPairDStream<String,Integer> mapToPair = flatMap.mapToPair(new PairFunction<String, String, Integer>() { private static final long serialVersionUID = 1L; @Override public Tuple2<String, Integer> call(String arg0) throws Exception { return new Tuple2<String, Integer>(arg0,1); } }); /** * 每隔5秒，計算最近60秒內的資料，那麼這個視窗大小就是60秒，裡面有12個rdd，在沒有計算之前，這些rdd是不會進行計算的。 * 那麼在計算的時候會將這12個rdd聚合起來，然後一起執行reduceByKeyAndWindow操作 ， * reduceByKeyAndWindow是針對視窗操作的而不是針對DStream操作的。 */// JavaPairDStream<String, Integer> reduceByKeyAndWindow = // mapToPair.reduceByKeyAndWindow(new Function2<Integer, Integer, Integer>() {// private static final long serialVersionUID = 1L;// @Override// public Integer call(Integer arg0, Integer arg1) throws Exception {// return arg0+arg1;// }// }, Durations.seconds(30),Durations.seconds(5)); /** * window視窗操作優化，設定加上後面10s的資料，然後減去統計在內的最前面10s的資料 * 可以看成只產生了兩個rdd，在計算之後，這些rdd才開始計算 */ JavaPairDStream<String, Integer> reduceByKeyAndWindow = mapToPair.reduceByKeyAndWindow(new Function2<Integer, Integer, Integer>() { private static final long serialVersionUID = 1L; @Override public Integer call(Integer arg0, Integer arg1) throws Exception { return arg0+arg1; } }, new Function2<Integer, Integer, Integer>() { private static final long serialVersionUID = 1L; @Override public Integer call(Integer arg0, Integer arg1) throws Exception { return arg0-arg1; } //下面的兩個時間必須是前面設定的durations屬性的整數倍 },Durations.seconds(60),Durations.seconds(10)); reduceByKeyAndWindow.print(); jsc.start(); jsc.awaitTermination(); jsc.stop(false); jsc.close(); }}