成都创新互联网站制作重庆分公司

如何理解ApacheFlinkCDC原理与使用

如何理解Apache Flink CDC原理与使用,相信很多没有经验的人对此束手无策,为此本文总结了问题出现的原因和解决方法,通过这篇文章希望你能解决这个问题。

峄城ssl适用于网站、小程序/APP、API接口等需要进行数据传输应用场景,ssl证书未来市场广阔!成为创新互联的ssl证书销售渠道,可以享受市场价格4-6折优惠!如果有意向欢迎电话联系或者加微信:13518219792(备注:SSL证书合作)期待与您的合作!

CDC (Change Data Capture)

Flink在1.11版本中新增了CDC的特性,简称 改变数据捕获。名称来看有点乱,我们先从之前的数据架构来看CDC的内容。如何理解Apache Flink CDC原理与使用

以上是之前的MySQL binlog日志处理流程,例如canal监听binlog把日志写入到kafka中。而Apache Flink实时消费Kakfa的数据实现mysql数据的同步或其他内容等。拆分来说整体上可以分为以下几个阶段。

  1. mysql开启binlog
  2. canal同步binlog数据写入到kafka
  3. flink读取kakfa中的binlog数据进行相关的业务处理。

整体的处理链路较长,需要用到的组件也比较多。Apache Flink CDC可以直接从数据库获取到binlog供下游进行业务计算分析。简单来说链路会变成这样如何理解Apache Flink CDC原理与使用也就是说数据不再通过canal与kafka进行同步,而flink直接进行处理mysql的数据。节省了canal与kafka的过程。

Flink 1.11中实现了mysql-cdc与postgre-CDC,也就是说在Flink 1.11中我们可以直接通过Flink来直接消费mysql,postgresql的数据进行业务的处理。

 
使用场景
  • 数据库数据的增量同步
  • 数据库表之上的物理化视图
  • 维表join
  • 其他业务处理
  • ...
 
MySQL CDC 操作实践

首先需要保证mysql数据库开启了binlog。未开启请查阅相关资料进行binlog的启用。自建默认是不开启binlog的。

  1. 源表
DROP TABLE IF EXISTS `t_test`;
CREATE TABLE `t_test` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `ip` varchar(255) DEFAULT NULL,
  `size` bigint(20) DEFAULT NULL
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=183 DEFAULT CHARSET=utf8mb4;
 
  1. 添加mysql-cdc相关依赖

  com.alibaba.ververica
  flink-connector-mysql-cdc
  1.1.0
  compile

 
  1. 相关代码实现
def main(args: Array[String]): Unit = {

    val envSetting = EnvironmentSettings.newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val tableEnv = StreamTableEnvironment.create(env, envSetting)
    val sourceDDL =
      "CREATE TABLE test_binlog (" +
      "   id INT NOT NULl," +
      "   ip STRING," +
      "   size INT" +
      ") WITH (" +
      "'connector' = 'mysql-cdc'," +
      "'hostname' = 'localhost'," +
      "'port' = '3306'," +
      "'username' = 'root'," +
      "'password' = 'cain'," +
      "'database-name' = 'test'," +
      "'table-name' = 't_test'" +
      ")"

    // 输出目标表
    val sinkDDL =
      "CREATE TABLE test_sink (\n" +
        " ip STRING,\n" +
        " countSum BIGINT,\n" +
        " PRIMARY KEY (ip) NOT ENFORCED\n" +
        ") WITH (\n" +
        " 'connector' = 'print'\n" +
        ")"

    val exeSQL =
      "INSERT INTO test_sink " +
        "SELECT ip, COUNT(1) " +
        "FROM test_binlog " +
        "GROUP BY ip"

    tableEnv.executeSql(sourceDDL)

    tableEnv.executeSql(sinkDDL)

    val result = tableEnv.executeSql(exeSQL)
    result.print()
  }
 
  1. 启动flink job,并且插入数据
INSERT INTO `test`.`t_test`( `ip`, `size`) VALUES (UUID(), 1231231);
INSERT INTO `test`.`t_test`( `ip`, `size`) VALUES (UUID(), 1231231);
INSERT INTO `test`.`t_test`( `ip`, `size`) VALUES (UUID(), 1231231);
...
 

插入数据可直接在console中看到flink处理的结果

如何理解Apache Flink CDC原理与使用  

Apache Flink CDC的方式替代了之前的canal+kafka节点.直接通过sql的方式来实现对mysql数据的同步。

看完上述内容,你们掌握如何理解Apache Flink CDC原理与使用的方法了吗?如果还想学到更多技能或想了解更多相关内容,欢迎关注创新互联行业资讯频道,感谢各位的阅读!


分享文章:如何理解ApacheFlinkCDC原理与使用
分享路径:http://cxhlcq.com/article/jdsoje.html

其他资讯

在线咨询

微信咨询

电话咨询

028-86922220(工作日)

18980820575(7×24)

提交需求

返回顶部