Flink Batch SQL 1.10 實(shí)踐

Flink作為流批統(tǒng)一的計算框架，在1.10中完成了大量batch相關(guān)的增強(qiáng)與改進(jìn)。1.10可以說是第一個成熟的生產(chǎn)可用的Flink Batch SQL版本，它一掃之前Dataset的羸弱，從功能和性能上都有大幅改進(jìn)，以下我從架構(gòu)、外部系統(tǒng)集成、實(shí)踐三個方面進(jìn)行闡述。

架構(gòu)

Stack

首先來看下stack，在新的Blink planner中，batch也是架設(shè)在Transformation上的，這就意味著我們和Dataset完全沒有關(guān)系了：

1. 我們可以盡可能的和streaming復(fù)用組件，復(fù)用代碼，有同一套行為。
2. 如果想要Table/SQL的toDataset或者fromDataset，那就完全沒戲了。盡可能的在Table的層面來處理吧。
3. 后續(xù)我們正在考慮在DataStream上構(gòu)建BoundedStream，給DataStream帶來批處理的功能。

網(wǎng)絡(luò)模型

Batch模式就是在中間結(jié)果落盤，這個模式和典型的Batch處理是一致的，比如MapReduce/Spark/Tez。

Flink以前的網(wǎng)絡(luò)模型也分為Batch和Pipeline兩種，但是Batch模式只是支持上下游隔斷執(zhí)行，也就是說資源用量可以不用同時滿足上下游共同的并發(fā)。但是另外一個關(guān)鍵點(diǎn)是Failover沒有對接好，1.9和1.10在這方面進(jìn)行了改進(jìn)，支持了單點(diǎn)的Failover。

建議在Batch時打開：

jobmanager.execution.failover-strategy = region

為了避免重啟過于頻繁導(dǎo)致JobMaster太忙了，可以把重啟間隔提高：

restart-strategy.fixed-delay.delay = 30 s

Batch模式的好處有：

• 容錯好，可以單點(diǎn)恢復(fù)
• 調(diào)度好，不管多少資源都可以運(yùn)行
• 性能差，中間數(shù)據(jù)需要落盤，強(qiáng)烈建議開啟壓縮
  taskmanager.network.blocking-shuffle.compression.enabled = true

Batch模式比較穩(wěn)，適合傳統(tǒng)Batch作業(yè)，大作業(yè)。

Pipeline模式是Flink的傳統(tǒng)模式，它完全和Streaming作業(yè)用的是同一套代碼，其實(shí)社區(qū)里Impala和Presto也是類似的模式，純走網(wǎng)絡(luò)，需要處理反壓，不落盤，它主要的優(yōu)缺點(diǎn)是：

• 容錯差，只能全局重來
• 調(diào)度差，你得保證有足夠的資源
• 性能好，Pipeline執(zhí)行，完全復(fù)用Stream，復(fù)用流控反壓等功能。

有條件可以考慮開啟Pipeline模式。

調(diào)度模型

Flink on Yarn支持兩種模式，Session模式和Per job模式，現(xiàn)在已經(jīng)在調(diào)度層次高度統(tǒng)一了。

1. Session模式?jīng)]有最大進(jìn)程限制，當(dāng)有Job需要資源時，它就會去Yarn申請新資源，當(dāng)Session有空閑資源時，它就會給Job復(fù)用，所以它的模型和PerJob是基本一樣的。
2. 唯一的不同只是：Session模式可以跨作業(yè)復(fù)用進(jìn)程。

另外，如果想要更好的復(fù)用進(jìn)程，可以考慮加大TaskManager的超時釋放：
resourcemanager.taskmanager-timeout = 900000

資源模型

先說說并發(fā)：

1. 對Source來說：目前Hive的table是根據(jù)InputSplit來定需要多少并發(fā)的，它之后能Chain起來的Operators自然都是和source相同的并發(fā)。
2. 對下游網(wǎng)絡(luò)傳輸過后的Operators(Tasks)來說：除了一定需要單并發(fā)的Task來說，其它Task全部統(tǒng)一并發(fā)，由table.exec.resource.default-parallelism統(tǒng)一控制。

我們在Blink內(nèi)部實(shí)現(xiàn)了基于統(tǒng)計信息來推斷并發(fā)的功能，但是其實(shí)以上的策略在大部分場景就夠用了。

Manage內(nèi)存

目前一個TaskManager里面含有多個Slot，在Batch作業(yè)中，一個Slot里只能運(yùn)行一個Task (關(guān)閉SlotShare)。

對內(nèi)存來說，單個TM會把Manage內(nèi)存切分成Slot粒度，如果1個TM中有n個Slot，也就是Task能拿到1/n的manage內(nèi)存。

我們在1.10做了重大的一個改進(jìn)就是：Task中chain起來的各個operators按照比例來瓜分內(nèi)存，所以現(xiàn)在配置的算子內(nèi)存都是一個比例值，實(shí)際拿到的還要根據(jù)Slot的內(nèi)存來瓜分。

這樣做的一個重要好處是：

1. 不管當(dāng)前Slot有多少內(nèi)存，作業(yè)能都run起來，這大大提高了開箱即用。
2. 不管當(dāng)前Slot有多少內(nèi)存，Operators都會把內(nèi)存瓜分干凈，不會存在浪費(fèi)的可能。

當(dāng)然，為了運(yùn)行的效率，我們一般建議單個Slot的manage內(nèi)存應(yīng)該大于500MB。

另一個事情，在1.10后，我們?nèi)コ薕nHeap的manage內(nèi)存，所以只有off-heap的manage內(nèi)存。

外部系統(tǒng)集成

Hive

強(qiáng)烈推薦Hive Catalog + Hive，這也是目前批處理最成熟的架構(gòu)。在1.10中，除了對以前功能的完善以外，其它做了幾件事：

1. 多版本支持，支持Hive 1.X 2.X 3.X
2. 完善了分區(qū)的支持，包括分區(qū)讀，動態(tài)/靜態(tài)分區(qū)寫，分區(qū)統(tǒng)計信息的支持。
3. 集成Hive內(nèi)置函數(shù)，可以通過以下方式來load:
   a)TableEnvironment.loadModule("hiveModule",new HiveModule("hiveVersion"))
4. 優(yōu)化了ORC的性能讀，使用向量化的讀取方式，但是目前只支持Hive 2+版本，且要求列沒有復(fù)雜類型。有沒有進(jìn)行過優(yōu)化差距在5倍量級。

兼容Streaming Connectors

得益于流批統(tǒng)一的架構(gòu)，目前的流Connectors也能在batch上使用，比如HBase的Lookup和Sink、JDBC的Lookup和Sink、Elasticsearch的Sink，都可以在Batch無縫對接使用起來。

實(shí)踐

SQL-CLI

在1.10中，SQL-CLI也做了大量的改動，比如把SQL-CLI做了stateful，里面也支持了DDL，還支持了大量的DDL命令，給SQL-CLI暴露了很多TableEnvironment的能力，這讓用戶可以方便得多。后續(xù)，我們也需要對接JDBC的客戶端，讓用戶可以更好的對接外部工具。但是SQL-CLI仍然待繼續(xù)改進(jìn)，比如目前仍然只支持Session模式，不支持Per Job模式。

編程方式

TableEnvironment tEnv = TableEnvironment.create(EnvironmentSettings
  .newInstance()
  .useBlinkPlanner()
  .inBatchMode()
  .build());

老的BatchTableEnv因為綁定了Dataset，而且區(qū)分Java和Scala，是不干凈的設(shè)計方式，所以Blink planner只支持新的TableEnv。

TableEnv注冊的source, sink, connector, functions，都是temporary的，重啟之后即失效了。如果需要持久化的object，考慮使用HiveCatalog。

tEnv.registerCatalog(“hive”, hiveCatalog);
tEnv.useCatalog(“hive”);

可以通過tEnv.sqlQuery來執(zhí)行DML，這樣可以獲得一個Table，我們也通過collect來獲得小量的數(shù)據(jù)：

Table table = tEnv.sqlQuery(“SELECT COUNT(*) FROM MyTable”);
List<Row> results = TableUtils.collectToList(table);
System.out.println(results);

可以通過tEnv.sqlUpdate來執(zhí)行DDL，但是目前并不支持創(chuàng)建hive的table，只能創(chuàng)建Flink類型的table：

tEnv.sqlUpdate(
   "CREATE TABLE myResult (" +
      "  cnt BIGINT"
      ") WITH (" +
      "  'connector.type'='jdbc'," 
         ……
      ")");

可以通過tEnv.sqlUpdate來執(zhí)行insert語句，Insert到臨時表或者Catalog表中，比如insert到上面創(chuàng)建的臨時JDBC表中：

tEnv.sqlUpdate(“INSERT INTO myResult SELECT COUNT(*) FROM MyTable”);
tEnv.execute(“MyJob”);

當(dāng)結(jié)果表是Hive表時，可以使用Overwrite語法，也可以使用靜態(tài)Partition的語法，這需要打開Hive的方言：

tEnv.getConfig().setSqlDialect(SqlDialect.HIVE);

結(jié)語

目前Flink batch SQL仍然在高速發(fā)展中，但是1.10已經(jīng)是一個可用的版本了，它在功能上、性能上都有很大的提升，后續(xù)還有很多有意思的features，等待著大家一起去挖掘。

原文鏈接

本文為阿里云原創(chuàng)內(nèi)容，未經(jīng)允許不得轉(zhuǎn)載。