Spark2.3（三十四）：Spark Structured Streaming之withWaterMark和windows窗口是否可以实现最近一小时统计

WaterMark除了可以限定来迟数据范围，是否可以实现最近一小时统计？

WaterMark目的用来限定参数计算数据的范围：比如当前计算数据内max timestamp是12::00，waterMark限定数据分为是60 minutes，那么如果此时输入11:00之前的数据就会被舍弃不参与统计，视为来迟范围超出了60minutes限定范围。

那么，是否可以借助它实现最近一小时的数据统计呢？

代码示例：

package com.dx.streaming

import java.sql.Timestamp
import java.text.SimpleDateFormat

import org.apache.spark.sql.streaming.OutputMode
import org.apache.spark.sql.{Encoders, SparkSession}
import org.apache.log4j.{Level, Logger}

case class MyEntity(id: String, timestamp: Timestamp, value: Integer)

object Main {
  Logger.getLogger("org.apache.spark").setLevel(Level.WARN);
  Logger.getLogger("akka").setLevel(Level.ERROR);
  Logger.getLogger("kafka").setLevel(Level.ERROR);

  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().appName("test").master("local[*]").getOrCreate()
    val lines = spark.readStream.format("socket").option("host", "192.168.0.141").option("port", 19999).load()

    var sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
    import spark.implicits._
    lines.as(Encoders.STRING)
      .map(row => {
        val fields = row.split(",")
        MyEntity(fields(0), new Timestamp(sdf.parse(fields(1)).getTime), Integer.valueOf(fields(2)))
      })
      .createOrReplaceTempView("tv_entity")

    spark.sql("select id,timestamp,value from tv_entity")
      .withWatermark("timestamp", "60 minutes")
      .createOrReplaceTempView("tv_entity_watermark")

    val resultDf = spark.sql(
      s"""
         |select id,sum(value) as sum_value
         |from  tv_entity_watermark
         |group id
         |""".stripMargin)

    val query = resultDf.writeStream.format("console").outputMode(OutputMode.Update()).start()

    query.awaitTermination()
    query.stop()
  }
}

当通过nc -lk 19999中依次（每组输入间隔几秒时间即可）输入如下数据时：

1,2018-12-01 12:00:01,100
2,2018-12-01 12:00:01,100

1,2018-12-01 12:05:01,100
2,2018-12-01 12:05:01,100

1,2018-12-01 12:15:01,100
2,2018-12-01 12:15:01,100

1,2018-12-01 12:25:01,100
2,2018-12-01 12:25:01,100

1,2018-12-01 12:35:01,100
2,2018-12-01 12:35:01,100

1,2018-12-01 12:45:01,100
2,2018-12-01 12:45:01,100

1,2018-12-01 12:55:01,100
2,2018-12-01 12:55:01,100

1,2018-12-01 13:05:02,100
2,2018-12-01 13:05:02,100

1,2018-12-01 13:15:01,100
2,2018-12-01 13:15:01,100

发现最终统计结果为：

id  , sum_value
   ,
   ,  

而不是期望的

id  , sum_value
   ,
   ,  

既然是不能限定数据统计范围是60minutes，是否需要借助于窗口函数window就可以实现呢？

是否需要借助于watermark和窗口函数window就可以实现最近1小时数据统计呢？

    spark.sql("select id,timestamp,value from tv_entity")
      .withWatermark("timestamp", "60 minutes")
      .createOrReplaceTempView("tv_entity_watermark")

    val resultDf = spark.sql(
      s"""
         |select id,sum(value) as sum_value
         |from  tv_entity_watermark
         |group window(timestamp,'60 minutes','60 minutes'),id
         |""".stripMargin)

    val query = resultDf.writeStream.format("console").outputMode(OutputMode.Update()).start()

依然输入上边的测试数据，会发现超过1小时候数据会重新开辟（归零后重新统计）一个统计结果，而不是滚动的一小时统计。

就是把上边的测试数据分为了两组来分别统计：

第一组（小时）参与统计数据：

,-- ::,
,-- ::,

,-- ::,
,-- ::,

,-- ::,
,-- ::,

,-- ::,
,-- ::,

,-- ::,
,-- ::,

,-- ::,
,-- ::,

,-- ::,
,-- ::,

第二组（小时）参与统计数据：

,-- ::,
,-- ::,

,-- ::,
,-- ::,

猜测总结：

根据上边测试结果可以推出一个猜测结论：

在spark structured streaming中是不存储参数统计的数据的，只是对数据进行了maxTimestamp.avgTimestamp,minTimestamp存储，同时只是对数据的统计结果进行存储，下次再次触发统计时只是在原有的统计结果之上进行累加等操作，而参与统计的数据应该是没有存储，否则这类需求应该是可以实现。

但是以下代码尝试确实是可以实现，缺点太耗费资源：

 package com.dx.streaming

 import java.sql.Timestamp
 import java.text.SimpleDateFormat

 import org.apache.spark.sql.streaming.OutputMode
 import org.apache.spark.sql.{Encoders, SparkSession}
 import org.apache.log4j.{Level, Logger}

 case class MyEntity(id: String, timestamp: Timestamp, value: Integer)

 object Main {
   Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
   Logger.getLogger("akka").setLevel(Level.ERROR)
   Logger.getLogger("kafka").setLevel(Level.ERROR)

   def main(args: Array[String]): Unit = {
     val spark = SparkSession.builder().appName("test").master("local[*]").getOrCreate()
     val lines = spark.readStream.format("socket").option("host", "192.168.0.141").option("port", 19999).load()

     var sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
     import spark.implicits._
     lines.as(Encoders.STRING)
       .map(row => {
         val fields = row.split(",")
         MyEntity(fields(0), new Timestamp(sdf.parse(fields(1)).getTime), Integer.valueOf(fields(2)))
       })
       .createOrReplaceTempView("tv_entity")

     spark.sql("select id,timestamp,value from tv_entity")
       .withWatermark("timestamp", "60 minutes")
       .createOrReplaceTempView("tv_entity_watermark")

     var resultDf =  spark.sql(
       s"""
          |select id,min(timestamp) min_timestamp,max(timestamp) max_timestamp,sum(value) as sum_value
          |from tv_entity_watermark
          |group by window(timestamp,'3600 seconds','60 seconds'),id
          |""".stripMargin)

     val query = resultDf.writeStream.format("console").outputMode(OutputMode.Update()).start()

     query.awaitTermination()
     query.stop()
   }
 }

使用spark streaming把历史结果保存到内存中实现最近一小时统计：

pom.xml

        <!--Spark -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
            <version>2.2.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming_2.11</artifactId>
            <version>2.2.0</version>
        </dependency>

java code:

package com.dx.streaming;

import java.io.Serializable;
import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.log4j.Level;
import org.apache.log4j.LogManager;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;

public class Main {
    private static List<MyEntity> store = new ArrayList<MyEntity>();
    private static JavaStreamingContext jssc;

    public static void main(String[] args) throws Exception {
        // set log4j programmatically
        LogManager.getLogger("org.apache.spark").setLevel(Level.WARN);
        LogManager.getLogger("akka").setLevel(Level.ERROR);
        LogManager.getLogger("kafka").setLevel(Level.ERROR);

        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        //
        System.out.println(sdf.parse("2018-12-04 11:00:00").getTime() - sdf.parse("2018-12-04 10:00:00").getTime());

        SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("NetworkWordCount");
        JavaSparkContext sc = new JavaSparkContext(conf);
        // jssc = new JavaStreamingContext(conf, Durations.seconds(10));
        jssc = new JavaStreamingContext(sc, Durations.seconds(10));

        JavaReceiverInputDStream<String> lines = jssc.socketTextStream("192.168.0.141", 19999);

        JavaDStream<MyEntity> dStream = lines.map(new Function<String, MyEntity>() {
            private static final long serialVersionUID = 1L;

            public MyEntity call(String line) throws Exception {
                String[] fields = line.split(",");
                MyEntity myEntity = new MyEntity();
                myEntity.setId(Integer.valueOf(fields[0]));
                myEntity.setTimestamp(Timestamp.valueOf(fields[1]));
                myEntity.setValue(Long.valueOf(fields[2]));
                return myEntity;
            }
        });

        // 不确定是否必须repartition(1)，目的避免外边这层循环多次循环，确保只执行一次大循环。
        dStream.repartition(1).foreachRDD(new VoidFunction<JavaRDD<MyEntity>>() {
            public void call(JavaRDD<MyEntity> tItems) throws Exception {
                System.out.println("print...");
                tItems.foreach(new VoidFunction<MyEntity>() {
                    public void call(MyEntity t) throws Exception {
                        System.out.println(">>>>>>>>>>>>>" + t.toString());
                        store.add(t);
                        System.out.println(store.size());
                    }
                });

                System.out.println("@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@");
                for (MyEntity myEntity : store) {
                    System.out.println("++++++++++++++++++++++" + myEntity.toString());
                }

                // 第一步：从store中超過1小時之前的數據剔除；
                MyEntity first = store.get(0);
                MyEntity last = store.get(store.size() - 1);
                // 超過一小時（这里为什么这么做，假设数据本身就是按照时间循序有序插入的，实际业务中如果相同可以这样做）
                while (last.getTimestamp().getTime() - first.getTimestamp().getTime() > 3600000) {
                    store.remove(0);
                    first = store.get(0);
                }

                // 第二步：執行業務統計代碼
                Map<Integer, Long> statistics = new HashMap<Integer, Long>();
                for (MyEntity myEntity : store) {
                    if (false == statistics.containsKey(myEntity.getId())) {
                        statistics.put(myEntity.getId(), myEntity.getValue());
                    } else {
                        statistics.put(myEntity.getId(), myEntity.getValue() + statistics.get(myEntity.getId()));
                    }
                }

                // 第三步：将结果写入关系数据库
                System.out.println("#######################print result##########################");
                for (Map.Entry<Integer, Long> kv : statistics.entrySet()) {
                    System.out.println(kv.getKey() + "," + kv.getValue());
                }
            }
        });

        jssc.start(); // Start the computation
        jssc.awaitTermination(); // Wait for the computation to terminate
    }
}

class MyEntity implements Serializable {
    private final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    private int id;
    private Timestamp timestamp;
    private long value;

    public int getId() {
        return id;
    }

    public void setId(int id) {
        this.id = id;
    }

    public Timestamp getTimestamp() {
        return timestamp;
    }

    public void setTimestamp(Timestamp timestamp) {
        this.timestamp = timestamp;
    }

    public long getValue() {
        return value;
    }

    public void setValue(long value) {
        this.value = value;
    }

    @Override
    public String toString() {
        return getId() + "," + sdf.format(new Date(getTimestamp().getTime())) + "," + getValue();
    }
}

输出日志

// :: INFO handler.ContextHandler: Started o.s.j.s.ServletContextHandler@62d73ead{/streaming/batch,null,AVAILABLE,@Spark}
// :: INFO handler.ContextHandler: Started o.s.j.s.ServletContextHandler@228cea97{/streaming/batch/json,null,AVAILABLE,@Spark}
// :: INFO handler.ContextHandler: Started o.s.j.s.ServletContextHandler@3db663d0{/static/streaming,null,AVAILABLE,@Spark}
// :: WARN storage.RandomBlockReplicationPolicy: Expecting  replicas with only  peer/s.
// :: WARN storage.BlockManager: Block input-- replicated to only  peer(s) instead of  peers
print...
>>>>>>>>>>>>>,-- ::,

>>>>>>>>>>>>>,-- ::,

@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
#######################print result##########################
,
,
// :: WARN storage.RandomBlockReplicationPolicy: Expecting  replicas with only  peer/s.
// :: WARN storage.BlockManager: Block input-- replicated to only  peer(s) instead of  peers
// :: WARN storage.RandomBlockReplicationPolicy: Expecting  replicas with only  peer/s.
// :: WARN storage.BlockManager: Block input-- replicated to only  peer(s) instead of  peers
print...
[Stage :>                                                          ( + ) / ]>>>>>>>>>>>>>,-- ::,

>>>>>>>>>>>>>,-- ::,

@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
#######################print result##########################
,
,
// :: WARN storage.RandomBlockReplicationPolicy: Expecting  replicas with only  peer/s.
// :: WARN storage.BlockManager: Block input-- replicated to only  peer(s) instead of  peers
// :: WARN storage.RandomBlockReplicationPolicy: Expecting  replicas with only  peer/s.
// :: WARN storage.BlockManager: Block input-- replicated to only  peer(s) instead of  peers
print...
[Stage :>                                                          ( + ) / ]>>>>>>>>>>>>>,-- ::,

>>>>>>>>>>>>>,-- ::,

@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
#######################print result##########################
,
,
// :: WARN storage.RandomBlockReplicationPolicy: Expecting  replicas with only  peer/s.
// :: WARN storage.BlockManager: Block input-- replicated to only  peer(s) instead of  peers
// :: WARN storage.RandomBlockReplicationPolicy: Expecting  replicas with only  peer/s.
// :: WARN storage.BlockManager: Block input-- replicated to only  peer(s) instead of  peers
print...
>>>>>>>>>>>>>,-- ::,

@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
#######################print result##########################
,
,
// :: WARN storage.RandomBlockReplicationPolicy: Expecting  replicas with only  peer/s.
// :: WARN storage.BlockManager: Block input-- replicated to only  peer(s) instead of  peers
// :: WARN storage.RandomBlockReplicationPolicy: Expecting  replicas with only  peer/s.
// :: WARN storage.BlockManager: Block input-- replicated to only  peer(s) instead of  peers
print...
>>>>>>>>>>>>>,-- ::,

>>>>>>>>>>>>>,-- ::,

>>>>>>>>>>>>>,-- ::,

@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
#######################print result##########################
,
,
// :: WARN storage.RandomBlockReplicationPolicy: Expecting  replicas with only  peer/s.
// :: WARN storage.BlockManager: Block input-- replicated to only  peer(s) instead of  peers
// :: WARN storage.RandomBlockReplicationPolicy: Expecting  replicas with only  peer/s.
// :: WARN storage.BlockManager: Block input-- replicated to only  peer(s) instead of  peers
print...
>>>>>>>>>>>>>,-- ::,

>>>>>>>>>>>>>,-- ::,

@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
#######################print result##########################
,
,
// :: WARN storage.RandomBlockReplicationPolicy: Expecting  replicas with only  peer/s.
// :: WARN storage.BlockManager: Block input-- replicated to only  peer(s) instead of  peers
// :: WARN storage.RandomBlockReplicationPolicy: Expecting  replicas with only  peer/s.
// :: WARN storage.BlockManager: Block input-- replicated to only  peer(s) instead of  peers
print...
>>>>>>>>>>>>>,-- ::,

>>>>>>>>>>>>>,-- ::,

@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
#######################print result##########################
,
,
// :: WARN storage.RandomBlockReplicationPolicy: Expecting  replicas with only  peer/s.
// :: WARN storage.BlockManager: Block input-- replicated to only  peer(s) instead of  peers
// :: WARN storage.RandomBlockReplicationPolicy: Expecting  replicas with only  peer/s.
// :: WARN storage.BlockManager: Block input-- replicated to only  peer(s) instead of  peers
print...
>>>>>>>>>>>>>,-- ::,

>>>>>>>>>>>>>,-- ::,

@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
#######################print result##########################
,
,
// :: WARN storage.RandomBlockReplicationPolicy: Expecting  replicas with only  peer/s.
// :: WARN storage.BlockManager: Block input-- replicated to only  peer(s) instead of  peers
// :: WARN storage.RandomBlockReplicationPolicy: Expecting  replicas with only  peer/s.
// :: WARN storage.BlockManager: Block input-- replicated to only  peer(s) instead of  peers
print...
>>>>>>>>>>>>>,-- ::,

>>>>>>>>>>>>>,-- ::,

@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
#######################print result##########################
,
,
print...
@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
++++++++++++++++++++++,-- ::,
#######################print result##########################
,
,