為了建立數(shù)據(jù)驅(qū)動(dòng)型企業(yè)，在數(shù)據(jù)目錄中實(shí)現(xiàn)企業(yè)數(shù)據(jù)資產(chǎn)的大眾化非常重要。利用統(tǒng)一的數(shù)據(jù)目錄，可以快速搜索數(shù)據(jù)集，并確定數(shù)據(jù)架構(gòu)、數(shù)據(jù)格式和位置。亞馬遜云科技Amazon Glue Data Catalog提供了一個(gè)統(tǒng)一的存儲(chǔ)庫，讓不同的系統(tǒng)能夠存儲(chǔ)和查找元數(shù)據(jù)，以跟蹤數(shù)據(jù)孤島中的數(shù)據(jù)。

Apache Flink是一個(gè)應(yīng)用廣泛的數(shù)據(jù)處理引擎，適用于可擴(kuò)展的流式處理ETL、分析和事件驅(qū)動(dòng)型應(yīng)用程序。該應(yīng)用程序提供具備容錯(cuò)能力的精確時(shí)間和狀態(tài)管理。Flink可以使用統(tǒng)一的API或應(yīng)用程序處理有界流（批處理）和無界流（流式處理）。在使用Apache Flink處理數(shù)據(jù)后，下游應(yīng)用程序可以使用統(tǒng)一的數(shù)據(jù)目錄訪問精心整理的數(shù)據(jù)。有了統(tǒng)一的元數(shù)據(jù)，數(shù)據(jù)處理和數(shù)據(jù)消耗應(yīng)用程序都可以使用相同的元數(shù)據(jù)訪問表。

如何將Amazon EMR中的Apache Flink與Amazon Glue Data Catalog集成，以便可以實(shí)時(shí)提取流數(shù)據(jù)，并近乎實(shí)時(shí)地訪問數(shù)據(jù)進(jìn)行業(yè)務(wù)分析。

?

Apache Flink連接器和目錄架構(gòu)

Apache Flink使用連接器和目錄，與數(shù)據(jù)和元數(shù)據(jù)進(jìn)行交互。下圖顯示了用于讀取/寫入數(shù)據(jù)的Apache Flink連接器以及用于讀取/寫入元數(shù)據(jù)的目錄架構(gòu)。

?

為了讀取/寫入數(shù)據(jù)，F(xiàn)link提供了用于讀取操作的DynamicTableSourceFactory接口以及用于寫入操作的DynamicTableSinkFactory接口。另有一個(gè)Flink連接器實(shí)施兩個(gè)接口，用于訪問不同存儲(chǔ)中的數(shù)據(jù)。例如，Flink FileSystem連接器提供了FileSystemTableFactory，用于在Hadoop Distributed File System（HDFS）或Amazon Simple Storage Service（Amazon S3）中讀取/寫入數(shù)據(jù)；Flink HBase連接器提供了HBase2DynamicTableFactory，用于在HBase中讀取/寫入數(shù)據(jù)；而Flink Kafka連接器提供了KafkaDynamicTableFactory，用于在Kafka中讀取/寫入數(shù)據(jù)。

對(duì)于元數(shù)據(jù)的讀取/寫入，F(xiàn)link提供了目錄接口。Flink有三種內(nèi)置的目錄實(shí)施。GenericInMemoryCatalog將目錄數(shù)據(jù)存儲(chǔ)在內(nèi)存中。JdbcCatalog將目錄數(shù)據(jù)存儲(chǔ)在JDBC支持的關(guān)系數(shù)據(jù)庫中。截至目前，JDBC目錄支持MySQL和PostgreSQL數(shù)據(jù)庫。HiveCatalog將目錄數(shù)據(jù)存儲(chǔ)在Hive Metastore中。HiveCatalog使用HiveShim來提供不同的Hive版本兼容性?？梢耘渲貌煌腗etastore客戶端，以使用Hive Metastore或Amazon Glue Data Catalog。

大多數(shù)Flink內(nèi)置連接器（例如Kafka、Amazon Kinesis、Amazon DynamoDB、Elasticsearch或FileSystem）都可以使用Flink?HiveCatalog，將元數(shù)據(jù)存儲(chǔ)在Amazon Glue Data Catalog中。然而，一些連接器實(shí)施（如Apache Iceberg）有單獨(dú)的目錄管理機(jī)制。Iceberg中的FlinkCatalog實(shí)施了Flink中的目錄接口。Iceberg中的FlinkCatalog對(duì)自己的目錄實(shí)施提供了封裝機(jī)制。下圖顯示了Apache Flink、Iceberg連接器和目錄之間的關(guān)系。

?

Apache Hudi也有自己的目錄管理功能。HoodieCatalog和HoodieHiveCatalog都在Flink中實(shí)施了目錄接口。HoodieCatalog將元數(shù)據(jù)存儲(chǔ)在諸如HDFS這樣的文件系統(tǒng)中。HoodieHiveCatalog將元數(shù)據(jù)存儲(chǔ)在Hive Metastore或Amazon Glue Data Catalog中，具體取決于是否將hive.metastore.client.factory.class配置為使用com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory。下圖顯示了Apache Flink、Hudi連接器和目錄之間的關(guān)系。

?

?

解決方案概覽

下圖顯示了本文章中描述的解決方案的整體架構(gòu)。

?

在此解決方案中，啟用Amazon RDS for MySQL?binlog來實(shí)時(shí)提取事務(wù)更改。Amazon EMR Flink CDC連接器讀取binlog數(shù)據(jù)并處理數(shù)據(jù)。經(jīng)過轉(zhuǎn)換的數(shù)據(jù)可以存儲(chǔ)在Amazon S3中。使用Amazon Glue Data Catalog來存儲(chǔ)元數(shù)據(jù)，例如表架構(gòu)和表位置。Amazon Athena或Amazon EMR Trino等下游數(shù)據(jù)使用者應(yīng)用程序訪問數(shù)據(jù)，以便用于業(yè)務(wù)分析。

下面是設(shè)置此解決方案的大致步驟：

1.?為Amazon RDS for MySQL啟用binlog并初始化數(shù)據(jù)庫。

2.?使用Amazon Glue Data Catalog創(chuàng)建EMR集群。

3.?在Amazon EMR中使用Apache Flink CDC提取更改數(shù)據(jù)捕獲（CDC,Change Data Capture）數(shù)據(jù)。

4.?將處理后的數(shù)據(jù)存儲(chǔ)在Amazon S3中，并將元數(shù)據(jù)存儲(chǔ)在Amazon Glue Data Catalog中。

5.?確認(rèn)所有表元數(shù)據(jù)都存儲(chǔ)在Amazon Glue Data Catalog中。

6.?通過Athena或Amazon EMR Trino使用數(shù)據(jù)進(jìn)行業(yè)務(wù)分析。

7.?更新和刪除Amazon RDS for MySQL中的源記錄，并驗(yàn)證數(shù)據(jù)湖表中是否發(fā)生相應(yīng)更改。

?

使用Amazon Glue Data Catalog

創(chuàng)建EMR集群

從Amazon EMR 6.9.0開始，F(xiàn)link表API/SQL可以與Amazon Glue Data Catalog集成。要使用Flink與Amazon Glue的集成，您必須創(chuàng)建Amazon EMR 6.9.0或更高版本。

1.?為Amazon EMR Trino與Data Catalog的集成創(chuàng)建文件iceberg.properties。當(dāng)表格格式為Iceberg時(shí)，您的文件應(yīng)包含如下內(nèi)容：

?

2. 將iceberg.properties上傳到S3存儲(chǔ)桶，例如DOC-EXAMPLE-BUCKET。

3. 創(chuàng)建trino-glue-catalog-setup.sh文件以配置Trino與Data Catalog的集成。使用trino-glue-catalog-setup.sh作為引導(dǎo)腳本。

4. 將trino-glue-catalog-setup.sh上傳到S3存儲(chǔ)桶（DOC-EXAMPLE-BUCKET）。

5. 創(chuàng)建flink-glue-catalog-setup.sh文件，以配置Flink與Data Catalog的集成。

6. 使用腳本運(yùn)行器，將flink-glue-catalog-setup.sh腳本作為步驟函數(shù)運(yùn)行。

7. 將flink-glue-catalog-setup.sh上傳到S3存儲(chǔ)桶（DOC-EXAMPLE-BUCKET）。

8. 使用Hive、Flink和Trino應(yīng)用程序創(chuàng)建EMR 6.9.0集群。

您可以使用Amazon命令行界面（Amazon CLI）或Amazon管理控制臺(tái)創(chuàng)建EMR集群。

?

小結(jié)

這篇文章展示了如何將Amazon EMR中的Apache Flink與Amazon Glue Data Catalog集成?？梢允褂肍link SQL連接器在不同的存儲(chǔ)區(qū)讀取/寫入數(shù)據(jù)，例如Kafka、CDC、HBase、Amazon S3、Iceberg或Hudi。您也可以將元數(shù)據(jù)存儲(chǔ)在Data Catalog中。Flink表API具有相同的連接器和目錄實(shí)施機(jī)制。在單個(gè)會(huì)話中，可以使用多個(gè)指向不同類型的目錄實(shí)例（如IcebergCatalog和HiveCatalog），然后在查詢中互換使用它們。也可以使用Flink表API編寫代碼，開發(fā)集成Flink與Data Catalog的相同解決方案。借助Amazon EMR Flink統(tǒng)一的批處理和流式數(shù)據(jù)處理功能，可以通過一個(gè)計(jì)算引擎提取和處理數(shù)據(jù)。通過將Apache Iceberg和Hudi集成到Amazon EMR中，可以構(gòu)建一個(gè)可演變且可擴(kuò)展的數(shù)據(jù)湖。通過Amazon Glue Data Catalog，可以統(tǒng)一管理所有企業(yè)數(shù)據(jù)目錄并輕松使用數(shù)據(jù)。