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

Question

我正在尝试使用两个不同的窗口聚合流并将其打印到控制台中.但是,只打印第一个流式查询.在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");

触发5s和10s聚合流的流式查询.不打印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为每个查询创建一个连接.您观察到的行为是因为"nc -lk"的工作原理.如果您使用netstat检查tcp连接,则在启动两个查询时会看到有两个连接.但是,"nc"仅将输入转发到一个连接.

我通过定义一个小实验来验证这种行为:首先,我创建了一个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后端服务器向每个打开的连接提供完全相同的数据,否则看起来不可能在由所提供的数据上定义多个聚合.