Apache Flink 是大資料領域又一新興框架。它與 Spark 的不同之處在於，它是使用流式處理來模擬批量處理的，因此能夠提供亞秒級的、符合 Exactly-once 語義的實時處理能力。Flink 的使用場景之一是構建實時的資料通道，在不同的儲存之間搬運和轉換資料。本文將介紹如何使用 Flink 開發實時 ETL 程式，並介紹 Flink 是如何保證其 Exactly-once 語義的。

讓我們來編寫一個從 Kafka 抽取資料到 HDFS 的程式。資料來源是一組事件日誌，其中包含了事件發生的時間，以時間戳的方式儲存。我們需要將這些日誌按事件時間分別存放到不同的目錄中，即按日分桶。時間日誌示例如下：

{"timestamp":1545184226.432,"event":"page_view","uuid":"ac0e50bf-944c-4e2f-bbf5-a34b22718e0c"}
{"timestamp":1545184602.640,"event":"adv_click","uuid":"9b220808-2193-44d1-a0e9-09b9743dec55"}
{"timestamp":1545184608.969,"event":"thumbs_up","uuid":"b44c3137-4c91-4f36-96fb-80f56561c914"}

產生的目錄結構為：
/user/flink/event_log/dt=20181219/part-0-1
/user/flink/event_log/dt=20181220/part-1-9

建立專案
Flink 應用程式需要使用 Java 8 編寫，我們可以使用 Maven 模板建立專案：
mvn archetype:generate \
-DarchetypeGroupId=org.apache.flink \
-DarchetypeArtifactId=flink-quickstart-java \
-DarchetypeVersion=1.7.0

將生成好的程式碼匯入到 IDE 中，可以看到名為 StreamingJob 的檔案，我們由此開始編寫程式。
Kafka 資料來源
Flink 對 Kafka 資料來源提供了原生支援，我們需要選擇正確的 Kafka 依賴版本，將其新增到 POM 檔案中：
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka-0.10_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
</dependency>

測試過程中，我們需要一個能夠執行的 Kafka 服務，讀者可以參照官方文件 搭建本地服務。在 Flink 中初始化 Kafka 資料來源時，傳入伺服器名和主題名就可以了：
Properties props = new Properties();
props.setProperty("bootstrap.servers", "localhost:9092");
FlinkKafkaConsumer010<String> consumer = new FlinkKafkaConsumer010<>(

"flink_test",newSimpleStringSchema(), props);

DataStream<String> stream = env.addSource(consumer);

Flink 會連線本地的 Kafka 服務，讀取 flink_test 主題中的資料，轉換成字串後返回。除了 SimpleStringSchema ，Flink 還提供了其他內建的反序列化方式，如 JSON、Avro 等，我們也可以編寫自定義邏輯。

流式檔案儲存
StreamingFileSink 替代了先前的 BucketingSink，用來將上游資料儲存到 HDFS 的不同目錄中。它的核心邏輯是分桶，預設的分桶方式是 DateTimeBucketAssigner，即按照處理時間分桶。處理時間指的是訊息到達 Flink 程式的時間，這點並不符合我們的需求。因此，我們需要自己編寫程式碼將事件時間從訊息體中解析出來，按規則生成分桶的名稱：

public class EventTimeBucketAssigner implements BucketAssigner<String, String> {
@Override
public String getBucketId(String element, Context context) {

JsonNode node = mapper.readTree(element);    
long date = (long) (node.path("timestamp").floatValue() * 1000);    
String partitionValue = new SimpleDateFormat("yyyyMMdd").format(new Date(date));    
return "dt=" + partitionValue;    

}
}

上述程式碼會使用 Jackson 庫對訊息體進行解析，將時間戳轉換成日期字串，新增字首後返回。如此一來，StreamingFileSink 就能知道應該將當前記錄放置到哪個目錄中了。
StreamingFileSink<String> sink = StreamingFileSink

.forRowFormat(new Path("/tmp/kafka-loader"), new SimpleStringEncoder<String>())    
.withBucketAssigner(new EventTimeBucketAssigner())    
.build();    

stream.addSink(sink);

forRowFormat 表示輸出的檔案是按行儲存的，對應的有 forBulkFormat，可以將輸出結果用 Parquet 等格式進行壓縮儲存。

關於 StreamingFileSink 還有一點要注意，它只支援 Hadoop 2.7 以上的版本，因為需要用到高版本檔案系統提供的 truncate 方法來實現故障恢復，這點下文會詳述。
開啟檢查點
程式碼編寫到這裡，其實已經可以通過 env.execute() 來運行了。但是，它只能保證 At-least-once 語義，即訊息有可能會被重複處理。要做到 Exactly-once，我們還需要開啟 Flink 的檢查點功能：
env.enableCheckpointing(60_000);
env.setStateBackend((StateBackend) new FsStateBackend("/tmp/flink/checkpoints"));
env.getCheckpointConfig().enableExternalizedCheckpoints(

ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION);

檢查點（Checkpoint）是 Flink 的故障恢復機制，同樣會在下文詳述。程式碼中，我們將狀態儲存方式由 MemoryStateBackend 修改為了 FsStateBackend，即使用外部檔案系統，如 HDFS，來儲存應用程式的中間狀態，這樣當 Flink JobManager 宕機時，也可以恢復過來。Flink 還支援 RocksDBStateBackend，用來存放較大的中間狀態，並能支援增量的狀態更新。

提交與管理指令碼
Flink 程式可以直接在 IDE 中除錯。我們也可以搭建一個本地的 Flink遊戲叢集，並通過 Flink CLI 命令列工具來提交指令碼：
bin/flink run -c com.shzhangji.flinksandbox.kafka.KafkaLoader target/flink-sandbox-0.1.0.jar

指令碼的執行狀態可以在 Flink 儀表盤中檢視：
640?wx_fmt=png

使用暫存點來停止和恢復指令碼
當需要暫停指令碼、或對程式邏輯進行修改時，我們需要用到 Flink 的暫存點機制（Savepoint）。暫存點和檢查點類似，同樣儲存的是 Flink 各個運算元的狀態資料（Operator State）。不同的是，暫存點主要用於人為的指令碼更替，而檢查點則主要由 Flink 控制，用來實現故障恢復。flink cancel -s 命令可以在停止指令碼的同時建立一個暫存點：
$ bin/flink cancel -s /tmp/flink/savepoints 1253cc85e5c702dbe963dd7d8d279038
Cancelled job 1253cc85e5c702dbe963dd7d8d279038. Savepoint stored in file:/tmp/flink/savepoints/savepoint-1253cc-0df030f4f2ee.

在 YARN 上執行
要將指令碼提交到 YARN 叢集上執行，同樣是使用 flink run 命令。首先將程式碼中指定檔案目錄的部分新增上 HDFS 字首，如 hdfs://localhost:9000/，重新打包後執行下列命令：
$ export HADOOP_CONF_DIR=/path/to/hadoop/conf
$ bin/flink run -m yarn-cluster -c com.shzhangji.flinksandbox.kafka.KafkaLoader target/flink-sandbox-0.1.0.jar
Submitted application application_1545534487726_0001

Flink 儀表盤會在 YARN Application Master 中執行，我們可以通過 ResourceManager 介面進入。返回的應用 ID 可以用來管理指令碼，新增 -yid 引數即可：
bin/flink cancel -s hdfs://localhost:9000/tmp/flink/savepoints -yid application_1545534487726_0001 84de00a5e193f26c937f72a9dc97f386

Flink 如何保證 Exactly-once 語義
Flink 實時處理程式可以分為三個部分，資料來源、處理流程、以及輸出。不同的資料來源和輸出提供了不同的語義保證，Flink 統稱為 聯結器。處理流程則能提供 Exactly-once 或 At-least-once 語義，需要看檢查點是否開啟。

實時處理與檢查點
Flink 的檢查點機制是基於 Chandy-Lamport 演算法的：Flink 會定時在資料流中安插輕量的標記資訊（Barrier），將訊息流切割成一組組記錄；當某個運算元處理完一組記錄後，就將當前狀態儲存為一個檢查點，提交給 JobManager，該組的標記資訊也會傳遞給下游；當末端的運算元（通常是 www.sangpi.comSink）處理完這組記錄並提交檢查點後，這個檢查點將被標記為“已完成”；當指令碼出現問題時，就會從最後一個“已完成”的檢查點開始重放記錄。

如果運算元有多個上游，Flink 會使用一種稱為“訊息對齊”的機制：如果某個上游出現延遲，當前運算元會停止從其它上游消費訊息，直到延遲的上游趕上進度，這樣就保證了運算元中的狀態不會包含下一批次的記錄。顯然，這種方式會引入額外的延遲，因此除了這種 EXACTLY_ONCE 模式，我們也可將檢查點配置為 AT_LEAST_ONCE，以獲得更高的吞吐量。具體方式請參考 官方文件。

可重放的資料來源
當出錯的指令碼需要從上一個檢查點恢復時，Flink 必須對資料進行重放，這就要求資料來源支援這一功能。Kafka 是目前使用得較多的訊息佇列，且支援從特定位點進行消費。具體來說，FlinkKafkaConsumer 類實現了 CheckpointedFunction 介面，會在檢查點中存放主題名、分割槽名、以及偏移量：

abstract class FlinkKafkaConsumerBase implements CheckpointedFunction {
public void initializeState(FunctionInitializationContext context) {

OperatorStateStore stateStore = context.getOperatorStateStore();    
this.unionOffsetStates = stateStore.getUnionListState(new ListStateDescriptor<>(    
    OFFSETS_STATE_NAME,    
    TypeInformation.of(new TypeHint<Tuple2<KafkaTopicPartition, Long>>() {})));    


if (context.isRestored()) {    
  for (Tuple2<KafkaTopicPartition, Long> kafkaOffset : unionOffsetStates.get()) {    
    restoredState.put(kafkaOffset.f0, kafkaOffset.f1);    
  }    
}    

}

public void snapshotState(FunctionSnapshotContext context) {

unionOffsetStates.clear();    
for (Map.Entry<KafkaTopicPartition, Long> kafkaTopicPartitionLongEntry : currentOffsets.entrySet()) {    
  unionOffsetStates.add(Tuple2.of(kafkaTopicPartitionLongEntry.getKey(),    
      kafkaTopicPartitionLongEntry.getValue()));    
}    

}
}

當資料來源運算元從檢查點或暫存點恢復時，我們可以在 TaskManager 的日誌中看到以下資訊，表明當前消費的偏移量是從運算元狀態中恢復出來的：
2018-12-23 10:56:47,380 INFO FlinkKafkaConsumerBase Consumer subtask 0 will start reading 2 partitions with offsets in restored state: {KafkaTopicPartition{topic='flink_test', partition=1}=725, KafkaTopicPartition{topic='flink_test', partition=0}=721}
1.

恢復寫入中的檔案
程式執行過程中，StreamingFileSink 首先會將結果寫入中間檔案，以 . 開頭、in-progress 結尾。這些中間檔案會在符合一定條件後更名為正式檔案，取決於使用者配置的 RollingPolicy，預設策略是基於時間（60 秒）和基於大小（128 MB）。當指令碼出錯或重啟時，中間檔案會被直接關閉；在恢復時，由於檢查點中儲存了中間檔名和成功寫入的長度，程式會重新開啟這些檔案，切割到指定長度（Truncate），然後繼續寫入。這樣一來，檔案中就不會包含檢查點之後的記錄了，從而實現 Exactly-once。

以 Hadoop 檔案系統舉例，恢復的過程是在 HadoopRecoverableFsDataOutputStream 類的建構函式中進行的。它會接收一個 HadoopFsRecoverable 型別的結構，裡面包含了中間檔案的路徑和長度。這個物件是 BucketState 的成員，會被儲存在檢查點中。
HadoopRecoverableFsDataOutputStream(FileSystem fs, HadoopFsRecoverable recoverable) {
this.tempFile = checkNotNull(recoverable.tempFile());
truncate(fs, tempFile, recoverable.offset());
out = fs.append(tempFile);
}

結論

Apache Flink 構建在實時處理之上，從設計之初就充分考慮了中間狀態的儲存，而且能夠很好地與現有 Hadoop 生態環境結合，因而在大資料領域非常有競爭力。它還在高速發展之中，近期也引入了 Table API、流式 SQL、機器學習等功能，像阿里巴巴這樣的公司也在大量使用和貢獻程式碼。Flink 的應用場景眾多，有很大的發展潛力，值得一試。