Flink CDC2Kafka 总结

前言

本文主要记录 CDC2Kafka 使用过程中的一些细节配置，以 mysql-cdc 为例。

背景需求

之前 CDC 的主要使用场景并没有写到 Kafka ，而是直接 Sink 到支持更新的目标端，比如 关系型数据库、Hudi、HBase，现在有了写到 Kafka 的需求，主要原因是有的项目的厂商不会使用数据湖表格式(Hudi、Iceberg、Delta Lake)，在使用 Hive 时不能支持ACID ，那么只能通过覆盖分区的方式实现更新删除，而覆盖更新对比小表来说可以通过定时全量抽取实现，对于大表每次都全量抽取效率太慢，所以最好可以通过增量变更日志+历史Hive表数据合并来实现，对于这种需求就得通过 CDC 格式来实现，CDC 格式有 debezium-json、canal-json、maxwell-json、ogg-json、changelog-json。 关系型数据库、Hudi、HBase等是不支持CDC格式的，常用的比较熟悉的组件就是Kafka，另外后面发现直接写 HDFS 文件也是可以的，本文先总结 CDC2Kafka+debezium-json 中的一些具体需求配置。

debezium-json

格式数据示例：

1
2
3
4
5
6
{"before":{"id":3,"name":"hudi3","price":3.33,"ts":2000,"dt":"20230331"},"after":null,"op":"d"}
{"before":null,"after":{"id":3,"name":"hudi4","price":3.33,"ts":2000,"dt":"20230331"},"op":"c"}

{"before":null,"after":{"id":4,"name":"hudi4","price":3.33,"ts":2000,"dt":"20230331"},"op":"c"}

{"before":{"id":4,"name":"hudi4","price":3.33,"ts":2000,"dt":"20230331"},"after":null,"op":"d"}

分别对应update、insert、delete ，其中 update 一次产生两条记录，先删除再插入。

more >>

展开全文 >>

Flink DataStream2Table 总结

前言

平时工作中有时会遇到 DataStream 转 Table 的需求，之前也写过几次，但是不总结就会忘掉还得专门扒拉之前写的代码，比较麻烦，现在总结一下，方便后面再有需要时查找。

DataStream(Tuple)

DataStream<Tuple> 转 Table

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
DataStream<Tuple5<Integer, String, Double, Long, String>> dataStream = env.fromCollection(
        java.util.Arrays.asList(
                Tuple5.of(1, "手机", 5999.99, 1733400000000L, "2025-12-05"),
                Tuple5.of(2, "电脑", 8999.50, 1733400001000L, "2025-12-05"),
                Tuple5.of(3, "耳机", 399.00, 1733400002000L, "2025-12-05")
        )
);

// 字段顺序必须和 Tuple5 一致：id(第1位)、name(第2位)、price(第3位)、ts(第4位)、dt(第5位)
Table table = tableEnv.fromDataStream(
        dataStream,
        "id, name, price, ts, dt" // 直接指定字段名，类型会自动匹配（Tuple5 的泛型）
);

table.execute().print();    
table.printSchema();

more >>

展开全文 >>

Hive 集群安装配置

前言

接Zookeeper 安装配置 和 Hadoop HA 集群安装配置，本文继续总结 Hive 集群安装配置。其实之前的文章中已经总结过部分，但是不全面：

前提

主机名修改、SSH互信、JDK安装等。

版本

Hive 3.1.3

安装包

https://archive.apache.org/dist/hive/hive-3.1.3/apache-hive-3.1.3-bin.tar.gz

more >>

展开全文 >>

Ray 集群安装配置

前言

作为新手，学习总结一下 Ray 的集群安装配置及简单使用。

官网

Ray 的核心特点

  • 简单的分布式编程模型：通过装饰器（@ray.remote）可以轻松将函数或类转换为分布式版本，无需深入了解分布式系统细节。
  • 自动任务调度：Ray 拥有高效的任务调度器，能够自动在集群中分配任务，优化资源利用率。
  • 支持状态 ful 计算：除了无状态的任务，Ray 还支持有状态的 Actor 模型，适合需要维护状态的场景（如训练模型、服务部署）。
  • 丰富的生态系统：包含多个上层库，如用于超参数调优的 Ray Tune、用于强化学习的 Ray RLlib、用于分布式训练的 Ray Train 等。
  • 灵活的部署方式：可在单机、集群、云环境（AWS、GCP、Azure 等）中部署，支持动态资源扩展。 more >>

展开全文 >>

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

2016年本科毕业于合肥工业大学<br><br>信息与计算科学专业<br><br>目前在济南一家公司从事大数据的工作~~