首先，看下window函数的图解：

下面这个代码是计算一分钟之内的单词数量统计，每两秒获取一次数据，同时处理数据时间也是两秒，窗口大小为1分钟

1.数据源

package com.ssm.test;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.ServerSocket;
import java.net.Socket;

public class SocketTest {

    public static void main(String[] args) {
        try{
            ServerSocket server=new ServerSocket(9999);
            System.out.println("The Server has started");
            
            Socket socket=server.accept();
            System.out.println("There was a socket client connected");
            
            BufferedReader br=new BufferedReader(new InputStreamReader(System.in));
            String line=br.readLine();
            PrintWriter writer=new PrintWriter(socket.getOutputStream());
            
            while(!line.equals("end")){
                writer = new PrintWriter(socket.getOutputStream());
                writer.println("If you are not brave enough, no one will back you up.");
                writer.flush();
                System.out.println(System.currentTimeMillis() + "\nServer send:\t"+line);
                Thread.sleep(2000);
            }
            writer.close();
            socket.close();
            server.close();
        }catch(Exception e) {
            e.printStackTrace();
        }
    }
}

2.spark streaming处理代码

package com.ssm.test;

import java.util.Arrays;
import java.util.Iterator;

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.Duration;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;

import scala.Tuple2;

public class WordsCountTest {

    public static void main(String[] args) throws Exception {
        
        SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("WordsCount");
        JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(2));
        JavaDStream<String> lines = jssc.socketTextStream("127.0.0.1", 9999).window(new Duration(60000));
        
        lines.flatMap(new FlatMapFunction<String, String>(){
            public Iterator<String> call(String x){
                return Arrays.asList(x.split(" ")).iterator();
            }
        }).mapToPair(new PairFunction<String, String, Integer>(){
            public Tuple2<String, Integer> call(String s){
                return new Tuple2<String, Integer>(s, 1);
            }
        }).reduceByKey(new Function2<Integer, Integer, Integer>(){
            public Integer call(Integer i1, Integer i2){
                return i1+i2;
            }
        }).print();
        
        jssc.start();
        jssc.awaitTermination();

    }

}