FlinkSQL 无缝对接 MySQL CDC 实时数据流入 Hive

2023-03-07 05:55:51

MySQL CDC 到 Hive 数据同步：使用 FlinkSQL 实现实时数据传输

背景介绍

随着数据量激增和业务需求日益复杂，我们经常需要将 MySQL 中的数据同步到其他系统，例如数据仓库或数据湖，以进行深入的数据分析。Hive 作为一种流行的大数据处理框架，可以轻松存储和处理海量数据，成为许多企业数据同步的目标之一。本文将介绍如何使用 Apache FlinkSQL 实现 MySQL CDC 数据到 Hive 的实时同步。

什么是 MySQL CDC？

MySQL CDC（变更数据捕获）是一种机制，用于跟踪和捕获 MySQL 数据库中数据的变更。启用 MySQL CDC 后，我们可以实时获取数据库中发生的所有数据变化，包括新增、修改和删除操作。这些变更数据可通过各种方式消费和处理，例如写入消息队列或数据仓库。

FlinkSQL 概述

Apache Flink 是一个开源的分布式流处理框架，用于处理大规模数据流。FlinkSQL 是 Flink 提供的 SQL 查询引擎，允许用户使用 SQL 语法进行数据查询和处理。FlinkSQL 具有强大的数据处理能力，可以轻松实现数据过滤、聚合、窗口计算等复杂操作。

系统架构

我们的数据同步系统架构如下：

MySQL 数据库：作为数据源，存储需要同步的数据。
Kafka：作为消息队列，用于存储从 MySQL CDC 中捕获的变更数据。
Flink：作为数据处理引擎，负责从 Kafka 中消费变更数据并写入 Hive。
Hive：作为数据仓库，存储从 MySQL 同步过来的数据，并提供 SQL 接口进行查询和分析。

实施步骤

1. 开启 MySQL CDC

binlog_row_image=FULL
binlog_format=ROW

2. 配置 Kafka

kafka-topics --create --topic mysql_cdc --partitions 1 --replication-factor 1

curl -X POST -H "Content-Type: application/json" --data '{
  "name": "mysql_cdc",
  "config": {
    "connector.class": "mysql_cdc",
    "tasks.max": "1",
    "database.hostname": "localhost",
    "database.port": "3306",
    "database.user": "root",
    "database.password": "password",
    "database.server.id": "1",
    "database.dbname": "my_database",
    "table.whitelist": "my_table"
  }
}' http://localhost:8083/connectors

3. 编写 FlinkSQL 作业

FlinkSQL作业1：

CREATE TABLE mysql_binlog (
  id INT,
  name STRING,
  age INT
) WITH (
  'connector' = 'kafka',
  'topic' = 'mysql_cdc',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'flink_cdc_group'
);

CREATE TABLE hive_table (
  id INT,
  name STRING,
  age INT
) WITH (
  'connector' = 'hive',
  'table-name' = 'my_hive_table'
);

INSERT INTO hive_table
SELECT id, name, age
FROM mysql_binlog;