• flink-cdc同步mysql数据到hive


    本文首发于我的个人博客网站 等待下一个秋-Flink

    什么是CDC

    CDC是(Change Data Capture 变更数据获取)的简称。核心思想是,监测并捕获数据库的变动(包括数据 或 数据表的插入INSERT、更新UPDATE、删除DELETE等),将这些变更按发生的顺序完整记录下来,写入到消息中间件中以供其他服务进行订阅及消费。

    Flink_CDC

    1. 环境准备

    • mysql

    • Hive

    • flink 1.13.5 on yarn

    说明:如果没有安装hadoop,那么可以不用yarn,直接用flink standalone环境吧。

    2. 下载下列依赖包

    下面两个地址下载flink的依赖包,放在lib目录下面。

    1. flink-sql-connector-hive-2.2.0_2.11-1.13.5.jar

    如果你的Flink是其它版本,可以来这里下载。

    说明:我hive版本是2.1.1,为啥这里我选择版本号是2.2.0呢,这是官方文档给出的版本对应关系:

    Metastore versionMaven dependencySQL Client JAR
    1.0.0 - 1.2.2flink-sql-connector-hive-1.2.2Download
    2.0.0 - 2.2.0flink-sql-connector-hive-2.2.0Download
    2.3.0 - 2.3.6flink-sql-connector-hive-2.3.6Download
    3.0.0 - 3.1.2flink-sql-connector-hive-3.1.2Download

    官方文档地址在这里,可以自行查看。

    3. 启动flink-sql client

    1. 先在yarn上面启动一个application,进入flink13.5目录,执行:
    bin/yarn-session.sh -d -s 2 -jm 1024 -tm 2048 -qu root.sparkstreaming -nm flink-cdc-hive
    
    • 1
    1. 进入flink sql命令行
    bin/sql-client.sh embedded -s flink-cdc-hive
    
    • 1

    img

    4. 操作Hive

    1) 首选创建一个catalog

    CREATE CATALOG hive_catalog WITH (
        'type' = 'hive',
        'hive-conf-dir' = '/etc/hive/conf.cloudera.hive'
    );
    
    • 1
    • 2
    • 3
    • 4

    这里需要注意:hive-conf-dir是你的hive配置文件地址,里面需要有hive-site.xml这个主要的配置文件,你可以从hive节点复制那几个配置文件到本台机器上面。

    2) 查询

    此时我们应该做一些常规DDL操作,验证配置是否有问题:

    use catalog hive_catalog;
    show databases;
    
    • 1
    • 2

    随便查询一张表

    use test
    show tables;
    select * from people;
    
    • 1
    • 2
    • 3

    可能会报错:

    image-20220915183211513

    把hadoop-mapreduce-client-core-3.0.0.jar放到flink的Lib目录下,这是我的,实际要根据你的hadoop版本对应选择。

    注意:很关键,把这个jar包放到Lib下面后,需要重启application,然后重新用yarn-session启动一个application,因为我发现好像有缓存,把这个application kill 掉,重启才行:

    image-20220915183454691

    然后,数据可以查询了,查询结果:

    image-20220915183102548

    5. mysql数据同步到hive

    mysql数据无法直接在flink sql导入hive,需要分成两步:

    1. mysql数据同步kafka;
    2. kafka数据同步hive;

    至于mysql数据增量同步到kafka,前面有文章分析,这里不在概述;重点介绍kafka数据同步到hive。

    1) 建表跟kafka关联绑定:

    前面mysql同步到kafka,在flink sql里面建表,connector=‘upsert-kafka’,这里有区别:

    CREATE TABLE product_view_mysql_kafka_parser(
    `id` int,
    `user_id` int,
    `product_id` int,
    `server_id` int,
    `duration` int,
    `times` string,
    `time` timestamp
    ) WITH (
     'connector' = 'kafka',
     'topic' = 'flink-cdc-kafka',
     'properties.bootstrap.servers' = 'kafka-001:9092',
     'scan.startup.mode' = 'earliest-offset',
     'format' = 'json'
    );
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15

    2) 建一张hive表

    创建hive需要指定SET table.sql-dialect=hive;,否则flink sql 命令行无法识别这个建表语法。为什么需要这样,可以看看这个文档Hive 方言

    -- 创建一个catalag用户hive操作
    CREATE CATALOG hive_catalog WITH (
        'type' = 'hive',
        'hive-conf-dir' = '/etc/hive/conf.cloudera.hive'
    );
    use catalog hive_catalog;
    
    -- 可以看到我们的hive里面有哪些数据库
    show databases;
    use test;
    show tables;
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11

    上面我们可以现在看看hive里面有哪些数据库,有哪些表;接下来创建一张hive表:

    CREATE TABLE product_view_kafka_hive_cdc (
      `id` int,
    `user_id` int,
    `product_id` int,
    `server_id` int,
    `duration` int,
    `times` string,
    `time` timestamp
    ) STORED AS parquet TBLPROPERTIES (
      'sink.partition-commit.trigger'='partition-time',
      'sink.partition-commit.delay'='0S',
      'sink.partition-commit.policy.kind'='metastore,success-file',
      'auto-compaction'='true',
      'compaction.file-size'='128MB'
    );
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15

    然后做数据同步:

    insert into hive_catalog.test.product_view_kafka_hive_cdc
    select * 
    from 
    default_catalog.default_database.product_view_mysql_kafka_parser;
    
    • 1
    • 2
    • 3
    • 4

    注意:这里指定表名,我用的是catalog.database.table,这种格式,因为这是两个不同的库,需要明确指定catalog - database - table。

    网上还有其它方案,关于mysql实时增量同步到hive:

    img

    网上看到一篇写的实时数仓架构方案,觉得还可以:

    image-20220916134859155

    参考资料

    https://nightlies.apache.org/flink/flink-docs-release-1.13/zh/docs/connectors/table/hive/hive_dialect/

  • 相关阅读:
    typescript核心
    创建InfluxDB数据库自定义函数
    【Python&GIS】基于Python实现栅格转面、面转栅格(栅格、矢量互转)
    ClickHouse技术研究及语法简介
    阿里专家精心整理分享的Java程序员面试笔试通关宝典PDF
    python实验课7~8---面向对象
    【Java基础】Math类、System类、toString方法、equals方法及冒泡排序实现
    OpenGL入门教程
    循环结构--do-while循环
    CUDA小白 - NPP(4) 图像处理 Data Exchange and Initialization(2)
  • 原文地址:https://blog.csdn.net/ddxygq/article/details/126889752