拋出疑無路？

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

示例代碼：

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

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

再現(xiàn)又一村！

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

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

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

• 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

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

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

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")
  }
}

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

Flink 的鍋？...

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

眾人拾柴

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

作者介紹

孫金城，51CTO社區(qū)編輯，Apache Flink PMC 成員，Apache Beam Committer，Apache IoTDB PMC 成員，ALC Beijing 成員，Apache ShenYu 導(dǎo)師，Apache 軟件基金會成員。關(guān)注技術(shù)領(lǐng)域流計算和時序數(shù)據(jù)存儲。