欧美人妇做爰免费视频,屄屄屄自拍视频,太粗大深好疼快拔出去软件,亚洲日韩免费电影,人妻3P真实偷拍,后入视频在线观看,内涵av入口,天天天日日

摘要：本文整理自云邪、雪盡在 Flink Forward Asia 2020 的分享，該分享以 4 個(gè)章節(jié)來詳細(xì)介紹如何利用 Flink SQL 構(gòu)建流批一體的 ETL 數(shù)據(jù)集成, 文章的主要內(nèi)容如下：

數(shù)據(jù)倉庫與數(shù)據(jù)集成
數(shù)據(jù)接入(E)
數(shù)據(jù)入倉/湖(L)
數(shù)據(jù)打?qū)?T)

Tips：點(diǎn)擊文末「閱讀原文」即可回顧作者原版分享視頻～

數(shù)據(jù)倉庫與數(shù)據(jù)集成

數(shù)據(jù)倉庫是一個(gè)集成的（Integrated），面向主題的（Subject-Oriented），隨時(shí)間變化的（Time-Variant），不可修改的（Nonvolatile）數(shù)據(jù)集合，用于支持管理決策。這是數(shù)據(jù)倉庫之父 Bill Inmon 在 1990 年提出的數(shù)據(jù)倉庫概念。該概念里最重要的一點(diǎn)就是“集成的”，其余特性都是一些方法論的東西。因?yàn)閿?shù)據(jù)倉庫首先要解決的問題，就是數(shù)據(jù)集成，就是將多個(gè)分散的、異構(gòu)的數(shù)據(jù)源整合在一起，消除數(shù)據(jù)孤島，便于后續(xù)的分析。這個(gè)不僅適用于傳統(tǒng)的離線數(shù)倉，也同樣適用于實(shí)時(shí)數(shù)倉，或者是現(xiàn)在火熱的數(shù)據(jù)湖。首先要解決的就是數(shù)據(jù)集成的問題。如果說業(yè)務(wù)的數(shù)據(jù)都在一個(gè)數(shù)據(jù)庫中，并且這個(gè)數(shù)據(jù)庫還能提供非常高效的查詢分析能力，那其實(shí)也用不著數(shù)據(jù)倉庫和數(shù)據(jù)湖上場(chǎng)了。

數(shù)據(jù)集成就是我們常稱作 ETL 的過程，分別是數(shù)據(jù)接入、數(shù)據(jù)清洗轉(zhuǎn)換打?qū)挕⒁约皵?shù)據(jù)的入倉入湖，分別對(duì)應(yīng)三個(gè)英文單詞的首字母，所以叫 ETL。ETL 的過程也是數(shù)倉搭建中最具工作量的環(huán)節(jié)。那么 Flink 是如何改善這個(gè) ETL 的過程的呢？我們先來看看傳統(tǒng)的數(shù)據(jù)倉庫的架構(gòu)。

傳統(tǒng)的數(shù)據(jù)倉庫，實(shí)時(shí)和離線數(shù)倉是比較割裂的兩套鏈路，比如實(shí)時(shí)鏈路通過 Flume和 Canal 實(shí)時(shí)同步日志和數(shù)據(jù)庫數(shù)據(jù)到 Kafka 中，然后在 Kafka 中做數(shù)據(jù)清理和打?qū)?。離線鏈路通過 Flume 和 Sqoop 定期同步日志和數(shù)據(jù)庫數(shù)據(jù)到 HDFS 和 Hive。然后在 Hive 里做數(shù)據(jù)清理和打?qū)挕?/span>

這里我們主要關(guān)注的是數(shù)倉的前半段的構(gòu)建，也就是到 ODS、DWD 層，我們把這一塊看成是廣義的 ETL 數(shù)據(jù)集成的范圍。那么在這一塊，傳統(tǒng)的架構(gòu)主要存在的問題就是這種割裂的數(shù)倉搭建這會(huì)造成很多重復(fù)工作，重復(fù)的資源消耗，并且實(shí)時(shí)、離線底層數(shù)據(jù)模型不一致，會(huì)導(dǎo)致數(shù)據(jù)一致性和質(zhì)量難以保障。同時(shí)兩個(gè)鏈路的數(shù)據(jù)是孤立的，數(shù)據(jù)沒有實(shí)現(xiàn)打通和共享。

那么 Flink 能給這個(gè)架構(gòu)帶來什么改變呢？

基于 Flink SQL 我們現(xiàn)在可以方便地構(gòu)建流批一體的 ETL 數(shù)據(jù)集成，與傳統(tǒng)數(shù)倉架構(gòu)的核心區(qū)別主要是這幾點(diǎn)：

Flink SQL 原生支持了 CDC 所以現(xiàn)在可以方便地同步數(shù)據(jù)庫數(shù)據(jù)，不管是直連數(shù)據(jù)庫，還是對(duì)接常見的 CDC工具。
Flink SQL 在最近的版本中持續(xù)強(qiáng)化了維表 join 的能力，不僅可以實(shí)時(shí)關(guān)聯(lián)數(shù)據(jù)庫中的維表數(shù)據(jù)，現(xiàn)在還能關(guān)聯(lián) Hive 和 Kafka 中的維表數(shù)據(jù)，能靈活滿足不同工作負(fù)載和時(shí)效性的需求。
基于 Flink 強(qiáng)大的流式 ETL 的能力，我們可以統(tǒng)一在實(shí)時(shí)層做數(shù)據(jù)接入和數(shù)據(jù)轉(zhuǎn)換，然后將明細(xì)層的數(shù)據(jù)回流到離線數(shù)倉中。
現(xiàn)在 Flink 流式寫入 Hive，已經(jīng)支持了自動(dòng)合并小文件的功能，解決了小文件的痛苦。

所以基于流批一體的架構(gòu)，我們能獲得的收益：

統(tǒng)一了基礎(chǔ)公共數(shù)據(jù)
保障了流批結(jié)果的一致性
提升了離線數(shù)倉的時(shí)效性
減少了組件和鏈路的維護(hù)成本

接下來我們會(huì)針對(duì)這個(gè)架構(gòu)中的各個(gè)部分,結(jié)合場(chǎng)景案例展開進(jìn)行介紹，包括數(shù)據(jù)接入，數(shù)據(jù)入倉入湖，數(shù)據(jù)打?qū)挕?/span>

數(shù)據(jù)接入

現(xiàn)在數(shù)據(jù)倉庫典型的數(shù)據(jù)來源主要來自日志和數(shù)據(jù)庫，日志接入現(xiàn)階段已經(jīng)非常成熟了，也有非常豐富的開源產(chǎn)品可供選擇，包括 Flume，F(xiàn)ilebeat，Logstash 等等都能很方便地采集日志到 Kafka 。這里我們就不作過多展開。

數(shù)據(jù)庫接入會(huì)復(fù)雜很多，常見的幾種 CDC 同步工具包括 Canal，Debezium，Maxwell。Flink 通過 CDC format 與這些同步工具做了很好的集成，可以直接消費(fèi)這些同步工具產(chǎn)生的數(shù)據(jù)。同時(shí) Flink 還推出了原生的 CDC connector，直連數(shù)據(jù)庫，降低接入門檻，簡(jiǎn)化數(shù)據(jù)同步流程。

我們先來看一個(gè)使用 CDC format 的例子?，F(xiàn)在常見的方案是通過 Debezium 或者 Canal 去實(shí)時(shí)采集 MySQL 數(shù)據(jù)庫的 binlog，并將行級(jí)的變更事件同步到 Kafka 中供 Flink 分析處理。在 Flink 推出 CDC format 之前，用戶要去消費(fèi)這種數(shù)據(jù)會(huì)非常麻煩，用戶需要了解 CDC 工具的數(shù)據(jù)格式，將 before，after 等字段都聲明出來，然后用 ROW_NUMBER 做個(gè)去重，來保證實(shí)時(shí)保留最后一行的語義。但這樣使用成本很高，而且也不支持 DELETE 事件。

現(xiàn)在 Flink 支持了 CDC format，比如這里我們?cè)?with 參數(shù)中可以直接指定 format = ‘debezium-json’，然后 schema 部分只需要填數(shù)據(jù)庫中表的 schema 即可。Flink 能自動(dòng)識(shí)別 Debezium 的 INSERT/UPDATE/DELETE 事件，并轉(zhuǎn)成 Flink 內(nèi)部的 INSERT/UPDATE/DELETE 消息。之后用戶可以在該表上直接做聚合、join 等操作，就跟操作一個(gè) MySQL 實(shí)時(shí)物化視圖一樣，非常方便。

在 Flink 1.12 版本中，F(xiàn)link 已經(jīng)原生支持了大部分常見的 CDC format，比如 Canal json、Debezium json、Debezium avro、Maxwell 等等。同時(shí) Flink 也開放了 CDC format 的接口，用戶可以實(shí)現(xiàn)自己的 CDC format 插件來對(duì)接自己公司的同步工具。

除此之外，F(xiàn)link 內(nèi)部原生支持了 CDC 的語義，所以可以很自然地直接去讀取 MySQL 的 binlog 數(shù)據(jù)并轉(zhuǎn)成 Flink 內(nèi)部的變更消息。所以我們推出了 MySQL CDC connector，你只需要在 with 參數(shù)中指定 connector=mysql-cdc，然后 select 這張表就能實(shí)時(shí)讀取 MySQL 中的全量 +CDC 增量數(shù)據(jù)，無需部署其他組件和服務(wù)。你可以把 Flink 中定義的這張表理解成是 MySQL 的實(shí)時(shí)物化視圖，所以在這張表上的聚合、join 等結(jié)果，跟實(shí)時(shí)在 MySQL 中運(yùn)行出來的結(jié)果是一致的。相比于剛剛介紹的 Debezium，Canal 的架構(gòu)，CDC connector 在使用上更加簡(jiǎn)單易用了，不用再去學(xué)習(xí)和維護(hù)額外組件，數(shù)據(jù)不需要經(jīng)過 Kafka 落地，減少了端到端延遲。而且支持先讀取全量數(shù)據(jù)，并無縫切換到 CDC 增量讀取上，也就是我們說的是流批一體，流批融合的架構(gòu)。

我們發(fā)現(xiàn) MySQL CDC connector 非常受用戶的歡迎，尤其是結(jié)合 OLAP 引擎，可以快速構(gòu)建實(shí)時(shí) OLAP 架構(gòu)。實(shí)時(shí) OLAP 架構(gòu)的一個(gè)特點(diǎn)就是將數(shù)據(jù)庫數(shù)據(jù)同步到 OLAP 中做即席查詢，這樣就無需離線數(shù)倉了。

以前是怎么做的呢？

之前用戶一般先用 datax 做個(gè)全量同步，然后用 canal 同步實(shí)時(shí)增量到 Kafka，然后從 Kafka 同步到 OLAP，這種架構(gòu)比較復(fù)雜，鏈路也很長?，F(xiàn)在很多公司都在用 Flink+ClickHouse 來快速構(gòu)建實(shí)時(shí) OLAP 架構(gòu)。我們只需要在 Flink 中定義一個(gè) mysql-cdc source，一個(gè) ClickHouse sink，然后提交一個(gè) insert into query 就完成了從 MySQL 到 ClickHouse 的實(shí)時(shí)同步工作，非常方便。而且，ClickHouse 有一個(gè)痛點(diǎn)就是 join 比較慢，所以一般我們會(huì)把 MySQL 數(shù)據(jù)打成一張大的明細(xì)寬表數(shù)據(jù)，再寫入 ClickHouse。這個(gè)在 Flink 中一個(gè) join 操作就完成了。而在 Flink 提供 MySQL CDC connector 之前，要在全量+增量的實(shí)時(shí)同步過程中做 join 是非常麻煩的。

當(dāng)然，這里我們也可以把 ClickHouse 替換成其他常見的 OLAP 引擎，比如阿里云的 Hologres。我們發(fā)現(xiàn)在阿里云上有很多的用戶都采用了這套鏈路和架構(gòu)，因?yàn)樗梢允〉魯?shù)據(jù)同步服務(wù)和消息中間件的成本，對(duì)于很多中小公司來說，在如今的疫情時(shí)代，控制成本是非常重要的。

當(dāng)然，這里也可以使用其他 OLAP 引擎，比如 TiDB。TiDB 官方也在最近發(fā)過一篇文章介紹這種 Flink+TiDB 的實(shí)時(shí) OLAP架構(gòu)。

數(shù)據(jù)入倉湖

剛剛我們介紹了基于 Flink SQL 可以非常方便地做數(shù)據(jù)接入，也就是 ETL 的 Extract 的部分。接下來，我們介紹一下 Flink SQL 在數(shù)據(jù)入倉入湖方面的能力，也就是 Load 的部分。

我們回顧下剛剛的流批一體的架構(gòu)圖，其中最核心的部分就是 Kafka 數(shù)據(jù)的流式入倉，正是這一流程打通了實(shí)時(shí)和離線數(shù)倉，統(tǒng)一了數(shù)倉的基礎(chǔ)公共數(shù)據(jù)，提升了離線數(shù)倉的時(shí)效性，所以我們針對(duì)這一塊展開講一講。

使用 Flink SQL 做流式數(shù)據(jù)入倉，非常的方便，而且 1.12 版本已經(jīng)支持了小文件的自動(dòng)合并，解決了小文件的痛點(diǎn)。可以看下右邊這段代碼，先在 Flink SQL 中使用 Hive dialect 創(chuàng)建一張 Hive 的結(jié)果表，然后通過 select from kafka 表 insert into Hive 表這樣一個(gè)簡(jiǎn)單 query，就可以提交任務(wù)實(shí)時(shí)將 Kafka 數(shù)據(jù)流式寫入 Hive。

如果要開啟小文件合并，只需要在 Hive 表參數(shù)中加上 auto-compaction = true，那么在流式寫入這張 Hive 表的時(shí)候就會(huì)自動(dòng)做小文件的 compaction。小文件合并的原理，是 Flink 的 streaming sink 會(huì)起一個(gè)小拓?fù)洌锩?temp writer 節(jié)點(diǎn)負(fù)責(zé)不斷將收到的數(shù)據(jù)寫入臨時(shí)文件中，當(dāng)收到 checkpoint 時(shí)，通知 compact coordinator 開始做小文件合并，compact coordinator 會(huì)將 compaction 任務(wù)分發(fā)給多個(gè) compact operator 并發(fā)地去做小文件合并。當(dāng) compaction 完成的時(shí)候，再通知 partition committer 提交整個(gè)分區(qū)文件可見。整個(gè)過程利用了 Flink 自身的 checkpoint 機(jī)制完成 compaction 的自動(dòng)化，無需起另外的 compaction 服務(wù)。這也是 Flink 流式入倉對(duì)比于其他入倉工具的一個(gè)核心優(yōu)勢(shì)。

除了流式入倉，F(xiàn)link 現(xiàn)在也支持流式入湖。以 Iceberg 舉例，基于 Iceberg 0.10，現(xiàn)在可以在 Flink SQL 里面直接 create 一個(gè) Iceberg catalog，在 Iceberg catalog 下可以 create table 直接創(chuàng)建 Iceberg表。然后提交 insert into query 就可以將流式數(shù)據(jù)導(dǎo)入到 Iceberg 中。然后在 Flink 中可以用 batch 模式讀取這張 Iceberg 表，做離線分析。不過 Iceberg 的小文件自動(dòng)合并功能目前還沒有發(fā)布，還在支持中。

剛剛介紹的是純 append 數(shù)據(jù)流式入倉入湖的能力，接下來介紹 CDC 數(shù)據(jù)流式入倉入湖的能力。我們先介紹 CDC 數(shù)據(jù)入 Kafka 實(shí)時(shí)數(shù)倉。其實(shí)這個(gè)需求在實(shí)時(shí)數(shù)倉的搭建中是非常常見的，比如同步數(shù)據(jù)庫 binlog 數(shù)據(jù)到 Kafka 中，又比如 join，聚合的結(jié)果是個(gè)更新流，用戶想把這個(gè)更新流寫到 Kafka 作為中間數(shù)據(jù)供下游消費(fèi)。

這在以前做起來會(huì)非常的麻煩，在 Flink 1.12 版本中，F(xiàn)link 引入了一個(gè)新的 connector ，叫做 upsert-kafka，原生地支持了 Kafka 作為一個(gè)高效的 CDC 流式存儲(chǔ)。

為什么說是高效的，因?yàn)榇鎯?chǔ)的形式是與 Kafka log compaction 機(jī)制高度集成的，Kafka 會(huì)對(duì) compacted topic 數(shù)據(jù)做自動(dòng)清理，且 Flink 讀取清理后的數(shù)據(jù)，仍能保證語義的一致性。而且像 Canal, Debezium 會(huì)存儲(chǔ) before,op_type 等很多無用的元數(shù)據(jù)信息，upsert-kafka 只會(huì)存儲(chǔ)數(shù)據(jù)本身的內(nèi)容，節(jié)省大量的存儲(chǔ)成本。使用上的話，只需要在 DDL 中聲明 connector = upsert-kafka，并定義 PK 即可。

比如我們這里定義了 MySQL CDC 的直播間表，以及一個(gè) upsert-kafka 的結(jié)果表，將直播間的數(shù)據(jù)庫同步到 Kafka 中。那么寫入 Kafka 的 INSERT 和 UPDATE 都是一個(gè)帶 key 的普通數(shù)據(jù)，DELETE 是一個(gè)帶 key 的 NULL 數(shù)據(jù)。Flink 讀取這個(gè) upsert-kafka 中的數(shù)據(jù)時(shí)，能自動(dòng)識(shí)別出 INSERT/UPDATE/DELETE 消息，消費(fèi)這張 upsert-kafka 表與消費(fèi) MySQL CDC 表的語義一致。并且當(dāng) Kafka 對(duì) topic 數(shù)據(jù)做了 compaction 清理后，F(xiàn)link 讀取清理后的數(shù)據(jù)，仍能保證語義的一致性。

CDC 數(shù)據(jù)入 Hive 數(shù)倉會(huì)麻煩一些，因?yàn)?Hive 本身不支持 CDC 的語義，現(xiàn)在的一種常見方式是先將 CDC 數(shù)據(jù)以 changelog-json 格式流式寫入到 HDFS。然后起個(gè) batch 任務(wù)周期性地將 HDFS 上的 CDC 數(shù)據(jù)按照 op 類型分為 INSERT, UPDATE, DELETE 三張表，然后做個(gè) batch merge。

數(shù)據(jù)打?qū)?/strong>

前面介紹了基于 Flink SQL 的 ETL 流程的 Extract 和 Load，接下來介紹 Transformation 中最常見的數(shù)據(jù)打?qū)挷僮鳌?/span>

數(shù)據(jù)打?qū)捠菙?shù)據(jù)集成中最為常見的業(yè)務(wù)加工場(chǎng)景，數(shù)據(jù)打?qū)捵钪饕氖侄尉褪?Join，F(xiàn)link SQL 提供了豐富的 Join 支持，包括 Regular Join、Interval Join、Temporal Join。

Regular Join 就是大家熟知的雙流 Join，語法上就是普通的 JOIN 語法。圖中案例是通過廣告曝光流關(guān)聯(lián)廣告點(diǎn)擊流將廣告數(shù)據(jù)打?qū)挘驅(qū)捄罂梢赃M(jìn)一步計(jì)算廣告費(fèi)用。從圖中可以看出，曝光流和點(diǎn)擊流都會(huì)存入 join 節(jié)點(diǎn)的 state，join 算子通過關(guān)聯(lián)曝光流和點(diǎn)擊流的 state 實(shí)現(xiàn)數(shù)據(jù)打?qū)?。Regular Join 的特點(diǎn)是，任意一側(cè)流都會(huì)觸發(fā)結(jié)果的更新，比如案例中的曝光流和點(diǎn)擊流。同時(shí) Regular Join 的語法與傳統(tǒng)批 SQL 一致，用戶學(xué)習(xí)門檻低。但需要注意的是，Regular join 通過 state 來存儲(chǔ)雙流已經(jīng)到達(dá)的數(shù)據(jù)，state 默認(rèn)永久保留，所以 Regular join 的一個(gè)問題是默認(rèn)情況下 state 會(huì)持續(xù)增長，一般我們會(huì)結(jié)合 state TTL 使用。

Interval Join 是一條流上需要有時(shí)間區(qū)間的 join，比如剛剛的廣告計(jì)費(fèi)案例中，它有一個(gè)非常典型的業(yè)務(wù)特點(diǎn)在里面，就是點(diǎn)擊一般發(fā)生在曝光之后的 10 分鐘內(nèi)。因此相對(duì)于 Regular Join，我們其實(shí)只需要關(guān)聯(lián)這10分鐘內(nèi)的曝光數(shù)據(jù)，所以 state 不用存儲(chǔ)全量的曝光數(shù)據(jù)，它是在 Regular Join 之上的一種優(yōu)化。要轉(zhuǎn)成一個(gè) Interval Join，需要在兩個(gè)流上都定義時(shí)間屬性字段（如圖中的 click_time 和 show_time）。并在 join 條件中定義左右流的時(shí)間區(qū)間，比如這里我們?cè)黾恿艘粋€(gè)條件：點(diǎn)擊時(shí)間需要大于等于曝光時(shí)間，同時(shí)小于等于曝光后 10 分鐘。與 Regular Join 相同， Interval Join 任意一條流都會(huì)觸發(fā)結(jié)果更新，但相比 Regular Join，Interval Join 最大的優(yōu)點(diǎn)是 state 可以自動(dòng)清理，根據(jù)時(shí)間區(qū)間保留數(shù)據(jù)，state 占用大幅減少。Interval Join 適用于業(yè)務(wù)有明確的時(shí)間區(qū)間，比如曝光流關(guān)聯(lián)點(diǎn)擊流，點(diǎn)擊流關(guān)聯(lián)下單流，下單流關(guān)聯(lián)成交流。

Temporal join (時(shí)態(tài)表關(guān)聯(lián)) 是最常用的數(shù)據(jù)打?qū)挿绞?，它常用來做我們熟知的維表 Join。在語法上，它需要一個(gè)顯式的 FOR SYSTEM_TIME AS OF 語句。它與 Regular Join 以及 Interval Join 最大的區(qū)別就是，維度數(shù)據(jù)的變化不會(huì)觸發(fā)結(jié)果更新，所以主流關(guān)聯(lián)上的維度數(shù)據(jù)不會(huì)再改變。Flink 支持非常豐富的 Temporal join 功能，包括關(guān)聯(lián) lookup DB，關(guān)聯(lián) changelog，關(guān)聯(lián) Hive 表。在以前，大家熟知的維表 join 一般都是關(guān)聯(lián)一個(gè)可以查詢的數(shù)據(jù)庫，因?yàn)榫S度數(shù)據(jù)在數(shù)據(jù)庫里面，但實(shí)際上維度數(shù)據(jù)可能有多種物理形態(tài)，比如 binlog 形式，或者定期同步到 Hive 中變成了 Hive 分區(qū)表的形式。在 Flink 1.12 中，現(xiàn)在已經(jīng)支持關(guān)聯(lián)這兩種新的維表形態(tài)。

Temporal Join Lookup DB 是最常見的維表 Join 方式，比如在用戶點(diǎn)擊流關(guān)聯(lián)用戶畫像的案例中，用戶點(diǎn)擊流在 Kafka 中，用戶實(shí)時(shí)畫像存放在 HBase 數(shù)據(jù)庫中，每個(gè)點(diǎn)擊事件通過查詢并關(guān)聯(lián) HBase 中的用戶實(shí)時(shí)畫像完成數(shù)據(jù)打?qū)?。Temporal Join Lookup DB 的特點(diǎn)是，維表的更新不會(huì)觸發(fā)結(jié)果的更新，維度數(shù)據(jù)存放在數(shù)據(jù)庫中，適用于實(shí)時(shí)性要求較高的場(chǎng)景，使用時(shí)我們一般會(huì)開啟 Async IO 和內(nèi)存 cache 提升查詢效率。

在介紹 Temporal Join Changelog 前，我們?cè)倏匆粋€(gè) Lookup DB 的例子，這是一個(gè)直播互動(dòng)數(shù)據(jù)關(guān)聯(lián)直播間維度的案例。這個(gè)案例中直播互動(dòng)數(shù)據(jù)（比如點(diǎn)贊、評(píng)論）存放在 Kafka 中，直播間實(shí)時(shí)的維度數(shù)據(jù)（比如主播、直播間標(biāo)題）存放在 MySQL 中，直播互動(dòng)的數(shù)據(jù)量是非常大的，為了加速訪問，常用的方案是加個(gè)高速緩存，比如把直播間的維度數(shù)據(jù)通過 CDC 同步，再存入 Redis 中，再做維表關(guān)聯(lián)。這種方案的問題是，直播的業(yè)務(wù)數(shù)據(jù)比較特殊，直播間的創(chuàng)建和直播互動(dòng)數(shù)據(jù)基本是同時(shí)產(chǎn)生的，因此互動(dòng)數(shù)據(jù)可能早早地到達(dá)了 Kafka 被 Flink 消費(fèi)，但是直播間的創(chuàng)建消息經(jīng)過了 Canal, Kafka，Redis, 這個(gè)鏈路比較長，數(shù)據(jù)延遲比較大，可能導(dǎo)致互動(dòng)數(shù)據(jù)查詢 Redis 時(shí)，直播間數(shù)據(jù)還未同步完成，導(dǎo)致關(guān)聯(lián)不上直播間數(shù)據(jù)，造成下游統(tǒng)計(jì)分析的偏差。

針對(duì)這類場(chǎng)景，F(xiàn)link 1.12 支持了 Temporal Join Changelog，通過從 changelog在 Flink state 中物化出維表來實(shí)現(xiàn)維表關(guān)聯(lián)。剛剛的場(chǎng)景有了更簡(jiǎn)潔的解決方案，我們可以通過 Flink CDC connector 把直播間數(shù)據(jù)庫表的 changelog 同步到 Kafka 中，注意我們看下右邊這段 SQL，我們用了 upsert-kafka connector 來將 MySQL binlog 寫入了 Kafka，也就是 Kafka 中存放了直播間變更數(shù)據(jù)的 upsert 流。然后我們將互動(dòng)數(shù)據(jù) temporal join 這個(gè)直播間 upsert 流，便實(shí)現(xiàn)了直播數(shù)據(jù)打?qū)挼墓δ堋?/span>

注意我們這里 FOR SYSTEM_TIME AS OF 不是跟一個(gè) processing time，而是左流的 event time，它的含義是去關(guān)聯(lián)這個(gè) event time 時(shí)刻的直播間數(shù)據(jù)，同時(shí)我們?cè)谥辈ラg upsert 流上也定義了 watermark，所以 temporal join changelog 在執(zhí)行上會(huì)做 watermark 等待和對(duì)齊，保證關(guān)聯(lián)上精確版本的結(jié)果，從而解決先前方案中關(guān)聯(lián)不上的問題。

我們?cè)敿?xì)解釋下 temporal join changelog 的過程，左流是互動(dòng)流數(shù)據(jù)，右流是直播間 changelog。直播間 changelog 會(huì)物化到右流的維表 state 中，state 相當(dāng)于一個(gè)多版本的數(shù)據(jù)庫鏡像，主流互動(dòng)數(shù)據(jù)會(huì)暫時(shí)緩存在左流的 state 中，等到 watermark 到達(dá)對(duì)齊后再去查維表 state 中的數(shù)據(jù)。比如現(xiàn)在互動(dòng)流和直播流的 watermark 都到了10:01分，互動(dòng)流的這條 10：01 分評(píng)論數(shù)據(jù)就會(huì)去查詢維表 state，并關(guān)聯(lián)上 103 房間的信息。當(dāng) 10：05 這條評(píng)論數(shù)據(jù)到來時(shí)，它不會(huì)馬上輸出，不然就會(huì)關(guān)聯(lián)上空的房間信息。它會(huì)一直等待，等到左右兩流的 watermark 都到 10：05 后，才會(huì)去關(guān)聯(lián)維表 state 中的數(shù)據(jù)并輸出。這個(gè)時(shí)候，它能關(guān)聯(lián)上準(zhǔn)確的 104 房間信息。

總結(jié)下，Temporal Join Changelog 的特點(diǎn)是實(shí)時(shí)性高，因?yàn)槭前凑?event time 做的版本關(guān)聯(lián)，所以能關(guān)聯(lián)上精確版本的信息，且維表會(huì)做 watermark 對(duì)齊等待，使得用戶可以通過 watermark 控制遲到的維表數(shù)。Temporal Join Changelog 中的維表數(shù)據(jù)都是存放在 temporal join 節(jié)點(diǎn)的 state 中，讀取非常高效，就像是一個(gè)本地的 Redis 一樣，用戶不再需要維護(hù)額外的 Redis 組件。

在數(shù)倉場(chǎng)景中，Hive 的使用是非常廣泛的，F(xiàn)link 與 Hive 的集成非常友好，現(xiàn)在已經(jīng)支持 Temporal Join Hive 分區(qū)表和非分區(qū)表。我們舉個(gè)典型的關(guān)聯(lián) Hive 分區(qū)表的案例：訂單流關(guān)聯(lián)店鋪數(shù)據(jù)。店鋪數(shù)據(jù)一般是變化比較緩慢的，所以業(yè)務(wù)方一般會(huì)按天全量同步店鋪表到 Hive 分區(qū)中，每天會(huì)產(chǎn)生一個(gè)新分區(qū)，每個(gè)分區(qū)是當(dāng)天全量的店鋪數(shù)據(jù)。

為了關(guān)聯(lián)這種 Hive 數(shù)據(jù)，只需我們?cè)趧?chuàng)建 Hive 分區(qū)表時(shí)指定右側(cè)這兩個(gè)紅圈中的參數(shù)，便能實(shí)現(xiàn)自動(dòng)關(guān)聯(lián) Hive 最新分區(qū)功能，partition.include = latestb 表示只讀取 Hive 最新分區(qū)，partition-name 表示選擇最新分區(qū)時(shí)按分區(qū)名的字母序排序。到 10 月 3 號(hào)的時(shí)候，Hive 中已經(jīng)產(chǎn)生了 10 月 2 號(hào)的新分區(qū), Flink 監(jiān)控到新分區(qū)后，就會(huì)重新加載10月2號(hào)的數(shù)據(jù)到 cache 中并替換掉10月1號(hào)的數(shù)據(jù)作為最新的維表。之后的訂單流數(shù)據(jù)關(guān)聯(lián)上的都是 cache 10 月 2 號(hào)分區(qū)的數(shù)據(jù)。Temporal join Hive 的特點(diǎn)是可以自動(dòng)關(guān)聯(lián) Hive 最新分區(qū)，適用于維表緩慢更新，高吞吐的業(yè)務(wù)場(chǎng)景。

總結(jié)一下我們剛剛介紹的幾種在數(shù)據(jù)打?qū)捴惺褂玫?join：

Regular Join 的實(shí)效性非常高，吞吐一般，因?yàn)?state 會(huì)保留所有到達(dá)的數(shù)據(jù)，適用于雙流關(guān)聯(lián)場(chǎng)景；
Interval Jon 的時(shí)效性非常好，吞吐較好，因?yàn)?state 只保留時(shí)間區(qū)間內(nèi)的數(shù)據(jù)，適用于有業(yè)務(wù)時(shí)間區(qū)間的雙流關(guān)聯(lián)場(chǎng)景；
Temporal Join Lookup DB 的時(shí)效性比較好，吞吐較差，因?yàn)槊織l數(shù)據(jù)都需要查詢外部系統(tǒng)，會(huì)有 IO 開銷，適用于維表在數(shù)據(jù)庫中的場(chǎng)景；
Temporal Join Changelog 的時(shí)效性很好，吞吐也比較好，因?yàn)樗鼪]有 IO 開銷，適用于需要維表等待，或者關(guān)聯(lián)準(zhǔn)確版本的場(chǎng)景；
Temporal Join Hive 的時(shí)效性一般，但吞吐非常好，因?yàn)榫S表的數(shù)據(jù)存放在cache 中，適用于維表緩慢更新的場(chǎng)景，高吞吐的場(chǎng)景。

總結(jié)

最后我們來總結(jié)下 Flink 在 ETL 數(shù)據(jù)集成上的能力。這是目前 Flink 數(shù)據(jù)集成的能力矩陣，我們將現(xiàn)有的外部存儲(chǔ)系統(tǒng)分為了關(guān)系型數(shù)據(jù)庫、KV 數(shù)據(jù)庫、消息隊(duì)列、數(shù)據(jù)湖、數(shù)據(jù)倉庫 5 種類型，可以從圖中看出 Flink 有非常豐富的生態(tài)，并且對(duì)每種存儲(chǔ)引擎都有非常強(qiáng)大的集成能力。

橫向上我們定義了 6 種能力，分別是 3 種數(shù)據(jù)接入能力：

全量讀取
流式讀取
CDC 流式讀取

一種數(shù)據(jù)打?qū)捘芰Γ?/span>

維度關(guān)聯(lián)；

以及兩種入倉/入湖能力：

流式寫入
CDC 寫入

可以看到 Flink 對(duì)各個(gè)系統(tǒng)的數(shù)據(jù)接入能力、維度打?qū)捘芰?、入倉/入湖能力都已經(jīng)非常完善了。在 CDC 流式讀取上，F(xiàn)link 已經(jīng)支持了主流的數(shù)據(jù)庫和 Kafka 消息隊(duì)列。在數(shù)據(jù)湖方向，F(xiàn)link 對(duì) Iceberg 的流式讀取和 CDC 寫入的功能也即將在接下來的 Iceberg 版本中發(fā)布。從這個(gè)能力矩陣可以看出，F(xiàn)link 的數(shù)據(jù)集成能力是非常全面的。

国产秋霞理论久久久电影-婷婷色九月综合激情丁香-欧美在线观看乱妇视频-精品国avA久久久久久久-国产乱码精品一区二区三区亚洲人-欧美熟妇一区二区三区蜜桃视频

基于 Flink SQL 構(gòu)建流批一體的 ETL 數(shù)據(jù)集成