Realtime Compute for Apache Flink：Realtime ComputeBlink產品公告（已停止新購）

Blink-2.2.7 是Blink-2.x系列中最新穩定版本，在Blink-1.x版本（最新穩定版本為Blink-1.6.4）進行了全面的升級，採用了自主研發的新一代儲存Niagara作為Statebackend的底層儲存，最佳化了SQL的效能，增加了一系列新功能：

• 主要特性

  • SQL

    • 新增Window Emit機制，可以控制Window結果的輸出策略，例如：1小時視窗，每1分鐘輸出一次。

    • 雙流Join支援miniBatch，針對不同情境最佳化了Retraction處理和State儲存結構，提高了效能。

    • AGG支援Filter文法，可以只彙總滿足條件的行。

    • 對Local-global AGG進行最佳化。

    • 重構了SQL的Optimize階段，解決了SQL編譯時間過長的問題。

    • SortedMapView中KEY支援多種資料類型：BOOLEAN、BYTE、SHORT、INT、LONG、FLOAT、DOUBLE、BIGDECIMAL、BIGINT、BYTE[]和STRING。

    • 最佳化了MIN、MAX、FIRST和LAST函數Retraction情境的效能。

    • 新增多種純量涵式，例如時區相關的解析TO_TIMESTAMP_TZ、格式化DATE_FORMAT_TZ和轉換函式CONVERT_TZ。

    • 對SQL和Connector模組的錯誤資訊進行了歸類，並對每種類型設計了相應的ERROR_CODE。

  • Connector

    • 支援使用者自訂的TableFactory註冊源表和結果表的Connector。

    • 支援使用者通過UDTF方式直接解析資料來源類型。

    • 支援讀取和寫入Kafka。

    • 支援寫入到Elasticsearch。

  • Runtime

    • 通過Blink Session機制，統一了使用者提交Job、擷取執行結果等行為。

    • 開放了調度外掛程式機制，允許計算模型根據需求自訂調度邏輯。

    • 在有限流的情況下，通過避免不必要的全圖重啟，提高了JobManager和Task FailOver的處理效率。

  • StateBackend

    • 使用NiagaraStateBackend替換RocksDBStateBackend，具備更好的讀寫效能。

    • (Experimental) NiagaraStateBackend支援計算儲存分離，支援Failover過程中State秒級恢複。

• 與Blink1.6.4不相容的文法

  功能項	影響	解決辦法
  TableFunction介面修改	所有使用自訂TableFunction的使用者。	更新代碼，實現新的getResultType介面。
  ScalarFunction介面修改	所有使用自訂ScalarFunction的使用者。	實現新的getResultType介面。
  AggregateFunction介面修改	所有使用自訂AggregateFunction的使用者。	實現新的getAccumulatorType和getResultType介面。例如，accumulator類型為Row(STRING, LONG)，Agg result的類型為STRING，則需要實現如下代碼。 public DataType getAccumulatorType\(\) { return DataTypes.createRowType\(DataTypes.String, DataTypes.LONG\); } public DataType getResultType\(\) { return DataTypes.String; }
  MapView建構函式修改	MapView建構函式形參類型由之前的TypeInformation變更為DataType。所以在自訂UDAF中聲明了MapView的Job都會受影響。	更新代碼，按DataType去構造MapView。例如MapView map = new MapView<>(Types.STRING, Types.INT);需要更新為MapView map = new MapView<>(DataTypes.STRING, DataTypes.INT);。
  當參數是LONG或INT時，除法和AVG傳回型別改為DOUBLE	以前的除法和AVG函數返回的是入參的欄位類型，現在是DOUBLE，會導致類型不符的錯誤。例如：除法和AVG的結果直接寫入結果表，可能會報結果表與Query欄位類型不符的錯誤。	在除法和AVG的結果上強制加上CAST。
  在比較BigDecimal 和Decimal資料類型的資料時，會考慮精度	用到Decimal的Job，可能會報BigDecimal類型不符的錯誤。	用到Decimal類型的Job，全域替換成帶精度的聲明方式，Decimal(38, 18)。
  NULL與字串的比較語義	1.x版本NULL與字串比較返回True，在2.x版本後遵循了SQL文法語義改為返回False。	所有NULL與字串比較的地方，例如：WHERE SPLIT_INDEX(shop_list, ':', 1) <> shop_id，如果SPLIT_INDEX(shop_list, ':', 1)返回了NULL，在1.x版本上WHERE條件會返回True，在2.x上會返回False，將資料過濾。

• 如何升級到Blink-2.x

  使用Blink-1.x版本的Job升級到Blink-2.x，需要進行資料回溯升級，資料回溯是指使用者根據業務需要，在啟動Job的時候，指定啟動位點，具體操作如下：

  1. 停止待升級Job（清除State）。

  2. 開發介面單擊右下角的Flink版本下拉式箭頭，修改Job的Blink版本為Blink-2.2.7，上線Job。

  3. 啟動修改後的Job並指定啟動位點。

     如果步驟3執行不成功，需要人工介入查明原因後，進行如下操作：

     • 快速修複SQL，重複步驟1、2、3。

     • 如果無法修複SQL，回退到原有Blink版本。

     • 如果無法產生Json Plan，可以嘗試設定如下參數：

       • blink.job.option.jmMemMB=4096

       • blink.job.submit.timeoutInSeconds=600

  Blink-2.0.1的UDX第三方外掛程式安裝包詳情，請參見Blink PDF文檔中的5.11.1章節。類似如下的異常，是因為UDX包的版本太低或者包衝突導致的。

  code:[30016], brief info:[get app plan failed], context info:[detail:[java.lang.NoClassDefFoundError: org/apache/flink/table/functions/aggfunctions/DoubleSumWithRetractAggFunction
  at java.lang.ClassLoader.defineClass1(Native Method)
  at java.lang.ClassLoader.defineClass(ClassLoader.java:788)
  at java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142)
  at java.net.URLClassLoader.defineClass(URLClassLoader.java:467)
  at java.net.URLClassLoader.access$100(URLClassLoader.java:73)