在 spark 结构化流中执行单独的流查询

Question

我正在尝试使用两个不同的 windows 聚合流并将其打印到控制台中。但是，只有第一个流式查询被打印出来。 tenSecsQ 不会打印到控制台中。

SparkSession spark = SparkSession
    .builder()
    .appName("JavaStructuredNetworkWordCountWindowed")
    .config("spark.master", "local[*]")
    .getOrCreate();

Dataset<Row> lines = spark
    .readStream()
    .format("socket")
    .option("host", host)
    .option("port", port)
    .option("includeTimestamp", true)
    .load();

Dataset<Row> words = lines
    .as(Encoders.tuple(Encoders.STRING(), Encoders.TIMESTAMP()))
    .toDF("word", "timestamp");

// 5 second window
Dataset<Row> fiveSecs = words
    .groupBy(
         functions.window(words.col("timestamp"), "5 seconds"),
         words.col("word")
    ).count().orderBy("window");

// 10 second window
Dataset<Row> tenSecs = words
    .groupBy(
          functions.window(words.col("timestamp"), "10 seconds"),
          words.col("word")
    ).count().orderBy("window");

触发 5 秒和 10 秒聚合流的流查询。不打印 10s 流的输出。只有 5s 被打印到控制台

// Start writeStream() for 5s window
StreamingQuery fiveSecQ = fiveSecs.writeStream()
    .queryName("5_secs")
    .outputMode("complete")
    .format("console")
    .option("truncate", "false")
    .start();

// Start writeStream() for 10s window
StreamingQuery tenSecsQ = tenSecs.writeStream()
    .queryName("10_secs")
    .outputMode("complete")
    .format("console")
    .option("truncate", "false")
    .start();

tenSecsQ.awaitTermination();

Answer 1

我一直在调查这个问题。

摘要：结构化流中的每个查询都使用 source 数据。套接字源为每个定义的查询创建一个新连接。在这种情况下看到的行为是因为 nc 仅将输入数据传送到第一个连接。

从今以后，除非我们可以确保连接的套接字源向每个打开的连接传递相同的数据，否则无法在套接字连接上定义多个聚合。

我在 Spark 邮件列表上讨论了这个问题。 Databricks 开发者 Shixiong Zhu 回答：

Spark creates one connection for each query. The behavior you observed is because how "nc -lk" works. If you use netstat to check the tcp connections, you will see there are two connections when starting two queries. However, "nc" forwards the input to only one connection.

我通过定义一个小实验验证了这个行为：首先，我创建了一个 SimpleTCPWordServer 来向每个打开的连接发送随机词，并创建了一个声明两个查询的基本结构化流作业。它们之间的唯一区别是第二个查询定义了一个额外的常量列来区分其输出：

val lines = spark
    .readStream
    .format("socket")
    .option("host", "localhost")
    .option("port", "9999")
    .option("includeTimestamp", true)
    .load()

val q1 = lines.writeStream
  .outputMode("append")
  .format("console")
  .trigger(Trigger.ProcessingTime("5 seconds"))
  .start()

val q2 = lines.withColumn("foo", lit("foo")).writeStream
  .outputMode("append")
  .format("console")
  .trigger(Trigger.ProcessingTime("7 seconds"))
  .start()

如果 StructuredStreaming 只使用一个流，那么我们应该会看到两个查询传递的是相同的词。如果每个查询都消耗一个单独的流，那么每个查询都会报告不同的词。

这是观察到的输出：

-------------------------------------------
Batch: 0
-------------------------------------------
+--------+-------------------+
|   value|          timestamp|
+--------+-------------------+
|champion|2017-08-14 13:54:51|
+--------+-------------------+

+------+-------------------+---+
| value|          timestamp|foo|
+------+-------------------+---+
|belong|2017-08-14 13:54:51|foo|
+------+-------------------+---+

-------------------------------------------
Batch: 1
-------------------------------------------
+-------+-------------------+---+
|  value|          timestamp|foo|
+-------+-------------------+---+
| agenda|2017-08-14 13:54:52|foo|
|ceiling|2017-08-14 13:54:52|foo|
|   bear|2017-08-14 13:54:53|foo|
+-------+-------------------+---+

-------------------------------------------
Batch: 1
-------------------------------------------
+----------+-------------------+
|     value|          timestamp|
+----------+-------------------+
|    breath|2017-08-14 13:54:52|
|anticipate|2017-08-14 13:54:52|
|   amazing|2017-08-14 13:54:52|
|    bottle|2017-08-14 13:54:53|
| calculate|2017-08-14 13:54:53|
|     asset|2017-08-14 13:54:54|
|      cell|2017-08-14 13:54:54|
+----------+-------------------+

我们可以清楚地看到每个查询的流是不同的。看起来不可能对 socket source 传送的数据定义多个聚合，除非我们可以保证 TCP 后端服务器向每个打开的连接传送完全相同的数据。

在 spark 结构化流中执行单独的流查询

Executing separate streaming queries in spark structured streaming

apache-spark

spark-structured-streaming