[發(fā)明專利]基于Flink實現(xiàn)Binlog到HIVE的實時采集方法和系統(tǒng)

在審

• 申請?zhí)枺?/td>: 202010729412.4
• 申請日: 2020-07-27
• 公開（公告）號: CN112100147A
• 公開（公告）日: 2020-12-18
• 發(fā)明（設計）人: 趙章萬
• 申請（專利權）人: 杭州玳數(shù)科技有限公司
• 主分類號: G06F16/18
• 分類號: G06F16/18;G06F16/182;G06F16/22;G06F16/2455;G06F16/25
• 代理公司: 常州佰業(yè)騰飛專利代理事務所(普通合伙) 32231
• 代理人: 陳紅橋

1.一種基于Flink實現(xiàn)Binlog到HIVE的實時數(shù)據(jù)采集方法，其特征在于，包括以下步驟：

建立所述Flink Job，并在所述Flink Job的初始化階段，配置并訂閱數(shù)據(jù)源的Binlog增量日志信息；

在所述Flink Job的數(shù)據(jù)讀取階段，對所述Binlog增量日志信息進行解析，以轉換為數(shù)據(jù)對象，其中，所述數(shù)據(jù)對象包括：數(shù)據(jù)庫信息和表名信息；

在所述Flink Job的數(shù)據(jù)寫入階段，根據(jù)所述數(shù)據(jù)庫信息和需要訂閱的Binlog的表名信息獲取表結構信息，并根據(jù)所述表結構信息創(chuàng)建HIVE分區(qū)表，以及獲取所述HIVE分區(qū)表對應的分布式系統(tǒng)下的存儲路徑，實現(xiàn)數(shù)據(jù)對分布式系統(tǒng)的直接寫入。

2.根據(jù)權利要求1所述的基于Flink實現(xiàn)Binlog到HIVE的實時數(shù)據(jù)采集方法，其特征在于，在所述Flink Job的初始化階段，還包括：

通過所述Flink Job的CheckPoint功能獲取上一次數(shù)據(jù)采集時保存的偏移量信息；

對所述Binlog日志訂閱的位置進行修正，以實現(xiàn)數(shù)據(jù)續(xù)傳。

3.根據(jù)權利要求1所述的基于Flink實現(xiàn)Binlog到HIVE的實時數(shù)據(jù)采集方法，其特征在于，在所述Flink Job的數(shù)據(jù)讀取階段，還包括：

通過所述Flink Job的CheckPoint記錄對所述Binlog增量日志信息進行解析時的日志消費的偏移量，以支持任務在下一次開始時進行數(shù)據(jù)續(xù)傳。

4.根據(jù)權利要求1所述的基于Flink實現(xiàn)Binlog到HIVE的實時數(shù)據(jù)采集方法，其特征在于，所述表結構信息包括：表字段內容、字段類型，所述數(shù)據(jù)庫信息包括：數(shù)據(jù)庫名和數(shù)據(jù)庫連接信息，所述方法還包括：

根據(jù)所述數(shù)據(jù)庫信息、所述表名信息、所述表結構信息對Binlog采集的所有表建立關聯(lián)關系，將同一數(shù)據(jù)庫名下相同表結構的表劃分為同一組。

5.根據(jù)權利要求1所述的基于Flink實現(xiàn)Binlog到HIVE的實時數(shù)據(jù)采集方法，其特征在于，所述根據(jù)所述表結構信息創(chuàng)建HIVE分區(qū)表，以及獲取所述HIVE分區(qū)表對應的分布式系統(tǒng)下的存儲路徑，實現(xiàn)數(shù)據(jù)對分布式系統(tǒng)的直接寫入，包括：

配置所述HIVE數(shù)據(jù)源信息，其中，所述HIVE數(shù)據(jù)源信息包括：HIVE JDBC連接信息、HIVE庫名、分區(qū)字段、HDFS配置信息；

通過所述HIVE JDBC創(chuàng)建HIVE分區(qū)表，并獲取分區(qū)表所對應的分布式文件系統(tǒng)的文件路徑；

根據(jù)待傳輸?shù)乃鰯?shù)據(jù)對象、所述HDFS配置信息、所述文件路徑，將所述數(shù)據(jù)寫入分布式文件系統(tǒng)中，其中，寫入的數(shù)據(jù)暫存于所述文件路徑下的臨時目錄之中。

6.根據(jù)權利要求5所述的基于Flink實現(xiàn)Binlog到HIVE的實時采集數(shù)據(jù)方法，其特征在于，

當所述CheckPoint功能被觸發(fā)時，將所述臨時目錄中暫存的數(shù)據(jù)對象轉移至所述HIVE分區(qū)表的文件路徑中。

7.一種基于Flink實現(xiàn)Binlog到HIVE的實時數(shù)據(jù)采集系統(tǒng)，其特征在于，包括：

建立組件，用于建立所述Flink Job，并在所述Flink Job的初始化階段，配置并訂閱數(shù)據(jù)源的Binlog增量日志信息；

數(shù)據(jù)讀取組件，用于在所述Flink Job的數(shù)據(jù)讀取階段，對所述Binlog增量日志信息進行解析，以轉換為數(shù)據(jù)對象，其中，所述數(shù)據(jù)對象包括：數(shù)據(jù)庫信息和表名信息；

分表管理組件，用于在所述Flink Job的數(shù)據(jù)寫入階段，根據(jù)所述數(shù)據(jù)庫信息和需要訂閱的Binlog的表名信息獲取表結構信息；

數(shù)據(jù)寫入組件，用于根據(jù)所述表結構信息創(chuàng)建HIVE分區(qū)表，以及獲取所述HIVE分區(qū)表對應的分布式系統(tǒng)下的存儲路徑，實現(xiàn)數(shù)據(jù)對分布式系統(tǒng)的直接寫入。

8.一種電子設備，其特征在于，包括：存儲器、處理器及存儲在所述存儲器上并可在所述處理器上運行的計算機程序，所述處理器執(zhí)行所述程序，以實現(xiàn)如權利要求1-6任一項所述的基于Flink實現(xiàn)Binlog到HIVE的實時數(shù)據(jù)采集方法。

9.一種非臨時性計算機可讀存儲介質，其上存儲有計算機程序，其特征在于，該程序被處理器執(zhí)行，以用于實現(xiàn)如權利要求1-6任一項所述的基于Flink實現(xiàn)Binlog到HIVE的實時數(shù)據(jù)采集方法。