Spark Structured Streaming——有状态流处理中带有窗口操作的事件处理

Question

我是Spark结构化流处理的新手，目前正在处理一个用例，其中结构化流应用程序将从Azure IoT中心-事件中心（例如每20秒）获取事件。

任务是使用这些事件并实时处理。为此，我在下面用Spark Java编写了Spark结构化流媒体程序。

以下是要点

1. 目前我已经应用了10分钟间隔和5分钟滑动窗口的窗口操作。
2. 水印被应用在以10分钟间隔的eventDate参数上。
3. 目前，我没有执行任何其他操作，只是将其存储在指定的位置，拼花格式。
4. 程序将一个事件存储在一个文件中。

问题：

1. 是否可以根据窗口时间以拼花格式将多个事件存储在一个文件中
2. 在这种情况下，窗口操作是如何工作的
3. 此外，我还想与之前的事件一起检查事件状态，并根据一些计算（例如，5分钟内未收到事件），我想更新状态

...

public class EventSubscriber {

   public static void main(String args[]) throws InterruptedException, StreamingQueryException {

    String eventHubCompatibleEndpoint = "<My-EVENT HUB END POINT CONNECTION STRING>";

    String connString = new ConnectionStringBuilder(eventHubCompatibleEndpoint).build();

    EventHubsConf eventHubsConf = new EventHubsConf(connString).setConsumerGroup("$Default")
            .setStartingPosition(EventPosition.fromEndOfStream()).setMaxRatePerPartition(100)
            .setReceiverTimeout(java.time.Duration.ofMinutes(10));

    SparkConf sparkConf = new SparkConf().setMaster("local[2]").setAppName("IoT Spark Streaming");

    SparkSession spSession = SparkSession.builder()
            .appName("IoT Spark Streaming")
            .config(sparkConf).config("spark.sql.streaming.checkpointLocation", "<MY-CHECKPOINT-LOCATION>")
            .getOrCreate();

    Dataset<Row> inputStreamDF = spSession.readStream()
            .format("eventhubs")
            .options(eventHubsConf.toMap())
            .load();

    Dataset<Row> bodyRow = inputStreamDF.withColumn("body", new Column("body").cast(DataTypes.StringType)).select("body");

    StructType jsonStruct = new StructType()
            .add("eventType", DataTypes.StringType)
            .add("payload", DataTypes.StringType);

    Dataset<Row> messageRow = bodyRow.map((MapFunction<Row, Row>) value -> {
        String valStr = value.getString(0).toString();

        String payload = valStr;

        Gson gson = new GsonBuilder().serializeNulls().setPrettyPrinting().create();

        JsonObject jsonObj = gson.fromJson(valStr, JsonObject.class);

        JsonElement methodName = jsonObj.get("method");

        String eventType = null;
        if(methodName != null) {
            eventType = "OTHER_EVENT";
        } else {
            eventType = "DEVICE_EVENT";
        }

        Row jsonRow = RowFactory.create(eventType, payload);
        return jsonRow;

    }, RowEncoder.apply(jsonStruct));

    messageRow.printSchema();

    Dataset<Row> deviceEventRowDS = messageRow.filter("eventType = 'DEVICE_EVENT'");

    deviceEventRowDS.printSchema();

    Dataset<DeviceEvent> deviceEventDS = deviceEventRowDS.map((MapFunction<Row, DeviceEvent>) value -> {

        String jsonString = value.getString(1).toString();

        Gson gson = new GsonBuilder().serializeNulls().setPrettyPrinting().create();

        DeviceMessage deviceMessage = gson.fromJson(jsonString, DeviceMessage.class);
        DeviceEvent deviceEvent = deviceMessage.getDeviceEvent();
        return deviceEvent;

    }, Encoders.bean(DeviceEvent.class));

    deviceEventDS.printSchema();

    Dataset<Row> messageDataset = deviceEventDS.select(
            functions.col("eventType"), 
            functions.col("deviceID"),
            functions.col("description"),
            functions.to_timestamp(functions.col("eventDate"), "yyyy-MM-dd hh:mm:ss").as("eventDate"),
            functions.col("deviceModel"),
            functions.col("pingRate"))
            .select("eventType", "deviceID", "description", "eventDate", "deviceModel", "pingRate");

    messageDataset.printSchema();

    Dataset<Row> devWindowDataset = messageDataset.withWatermark("eventDate", "10 minutes")
            .groupBy(functions.col("deviceID"),
                    functions.window(
                            functions.col("eventDate"), "10 minutes", "5 minutes"))
            .count();

    devWindowDataset.printSchema();

    StreamingQuery query = devWindowDataset.writeStream().outputMode("append")
            .format("parquet")
            .option("truncate", "false")
            .option("path", "<MY-PARQUET-FILE-OUTPUT-LOCATION>")
            .start();

    query.awaitTermination();
}}

...

任何关于这方面的帮助或指导都是有用的。

感谢和问候，

阿维纳什·德什穆克

Answer 1

是否可以根据窗口时间以拼花格式将多个事件存储在一个文件中？

是的。

在这种情况下，窗口操作是如何工作的？

以下代码是Spark结构化流媒体应用程序的主要部分：

Dataset<Row> devWindowDataset = messageDataset
  .withWatermark("eventDate", "10 minutes")
  .groupBy(
    functions.col("deviceID"),
    functions.window(functions.col("eventDate"), "10 minutes", "5 minutes"))
  .count();

也就是说，底层状态存储应该按照deviceID和eventDate将状态保持10分钟，对于延迟事件，额外的10分钟（按照带水印）。换句话说，在流式查询开始20分钟后，当一个事件eventDate20分钟后，您应该会看到结果出来。

with Watermark用于后期事件，因此即使group by将产生结果，结果也不会仅在超过水印阈值之前发出。

同样的程序每10分钟应用一次（10分钟水印），用5分钟的窗口幻灯片。

将带有窗口的groupBy操作符视为多列聚合。

此外，我还想与之前的事件一起检查事件状态，并根据一些计算（例如，5分钟内未收到事件），我想更新状态。

这听起来像是KeyValueGroupedDataset的一个用例。flatMapGroupsWithState运算符（又称任意有状态流聚合）。查阅任意有状态操作。

也可能只需要众多聚合标准函数中的一个或用户定义的聚合函数（UDAF）。