拋出疑無路？

【Flink 1.10】- 有一種情況是所有的系統或應用之間的橋梁都是Kafka，而這個時候恰恰是上游需要做Unbound的聚合統計。From @PyFlink 企業用戶。

示例代碼：

INSERT INTO kafkaSink 
    SELECT 
      id, 
      SUM(cnt) 
    FROM csvSource 
    GROUP BY id

執行這個SQL，在【Flink 1.10】版本會拋出如下異常：

再現又一村！

【Flink-1.10】這個問題是因Flink內部Retract機制導致，在沒有考慮對Chanage log全鏈路支持之前，無法在Kafka這樣的Append only的消息隊列增加對Retract/Upsert的支持。這個做法是出于語義完整性考慮做出的決定。但現實業務場景總是有著這樣或那樣的實際業務需求，業務不關心你語義是否okay，業務關心我不改變我原有的技術選型。

在這個基礎之上只要你告訴我Sink到Kafka的行為就行，我會根據你的產出行為，在業務上面做適配，所以這個時候就是實用為主，不管什么語義不語義了......，所以這個時候應該怎么辦呢？

我們的做法是將 Kafka的sink由原有的AppendStreamTableSink變成UpsertStreamTableSink或者RetractStreamTableSink。但出于性能考慮，我們改變成UpsertStreamTableSink，這個改動不多，但是對于初學者來講還是不太愿意動手改代碼，所以為大家提供一份：

• KafkaTableSinkBase.java
  

https://github.com/sunjincheng121/know_how_know_why/blob/master/QA/upsertKafka/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSinkBase.java

• KafkaTableSourceSinkFactoryBase.java

https://github.com/sunjincheng121/know_how_know_why/blob/master/QA/upsertKafka/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceSinkFactoryBase.java

在你的項目創建 org.apache.flink.streaming.connectors.kafka包 并把上面的兩個類放入該包，用于覆蓋官方KafkaConnector里面的實現。

特別強調：這樣的變化會導致寫入Kafka的結果不會是每個Group Key只有一條結果，而是每個Key可能有很多條結果。這個大家可以自行測試一下：

package cdc

import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.scala._

/**
 * Test for sink data to Kafka with upsert mode.
 */
object UpsertKafka {
  def main(args: Array[String]): Unit = {
    val sourceData = "file:///Users/jincheng.sunjc/work/know_how_know_why/QA/upsertKafka/src/main/scala/cdc/id_cnt_data.csv"
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val tEnv = StreamTableEnvironment.create(env)

    val sourceDDL = "CREATE TABLE csvSource (" +
      "  id VARCHAR," +
      "  cnt INT" +
      ") WITH (" +
      "'connector.type' = 'filesystem'," +
      "'connector.path' = '" + sourceData + "'," +
      "'format.type' = 'csv'" +
      ")"

    val sinkDDL = "CREATE TABLE kafkaSink (" +
      "  id VARCHAR," +
      "  cnt INT " +
      ") WITH (" +
      "'connector.type' = 'kafka'," +
      "'connector.version' = '0.10'," +
      "'connector.topic' = 'test'," +
      "'connector.properties.zookeeper.connect' = 'localhost:2181'," +
      "'connector.properties.bootstrap.servers' = 'localhost:9092'," +
      "'connector.properties.group.id' = 'data_Group'," +
      "'format.type' = 'json')"

    tEnv.sqlUpdate(sourceDDL)
    tEnv.sqlUpdate(sinkDDL)

    val sql = "INSERT INTO kafkaSink" +  
    " SELECT id, SUM(cnt) FROM csvSource GROUP BY id"
    tEnv.sqlUpdate(sql)
    env.execute("RetractKafka")
  }
}

當然，也可以clone我的git代碼【https://github.com/sunjincheng121/know_how_know_why/tree/master/QA/upsertKafka】直觀體驗一下。由于本系列文章只關注解決問題，不論述細節原理，有關原理性知識，我會在我的視頻課程《Apache 知其然，知其所以然》中進行介紹。

Flink 的鍋？...

看到上面的問題有些朋友可能會問，既然知道問題，知道有實際業務需求，為啥Flink不改進，不把這種情況支持掉呢？問的好，就這個問題而言，Flink是委屈的，Flink已經在努力支持這個場景了，預期Flink-1.12的版本大家會體驗到完整的CDC（change data capture）支持。

眾人拾柴

期待你典型問題的拋出... 我將知無不言...言無不盡... 我在又一村等你...

作者介紹

孫金城，51CTO社區編輯，Apache Flink PMC 成員，Apache Beam Committer，Apache IoTDB PMC 成員，ALC Beijing 成員，Apache ShenYu 導師，Apache 軟件基金會成員。關注技術領域流計算和時序數據存儲。