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", new SimpleStringSchema(), 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 的應用場景眾多，有很大的發展潛力，值得一試。