Flink用户自定义连接器（Table API Connectors）学习总结

前言

结合官网文档和自定义实现一个flink-http-connector，来学习总结Flink用户自定义连接器（Table API Connectors）。

背景

前段时间有个需求：需要Flink查询API接口，将返回的数据转为Flink Table，然后基于Table进行后面的计算。这个需求可以写Flink代码实现：使用HttpClient API请求接口返回数据，然后将返回的数据转为DataStream,最后再将DataStream转为Table。我想了一下是不是可以通过SQL的形式实现这种需求，于是在网上查了一下，还真有。Star比较多的项目：https://github.com/getindata/flink-http-connector.git,但是它要求Java 11，并且它的Http Source只支持Lookup Joins，限制太多，并不能满足我的需求。所以最终又尝试学习了自己写自定义的Table API Connectors，这样可以比较灵活的实现需求。

官网文档

官网文档地址：https://nightlies.apache.org/flink/flink-docs-release-1.17/zh/docs/dev/table/sourcessinks/,官网文档详细介绍了自定义连接器的概念、扩展点并给出了一个完整的代码实例，实例实现了自定义socket连接器。本文摘录一部分官方文档，便于自己理解。

动态表是 Flink Table & SQL API的核心概念，用于统一有界和无界数据的处理。

动态表只是一个逻辑概念，因此 Flink 并不拥有数据。相应的，动态表的内容存储在外部系统（ 如数据库、键值存储、消息队列 ）或文件中。

动态 sources 和动态 sinks 可用于从外部系统读取数据和向外部系统写入数据。在文档中，sources 和 sinks 常在术语连接器 下进行总结。

Flink 为 Kafka、Hive 和不同的文件系统提供了预定义的连接器。有关内置 table sources 和 sinks 的更多信息，请参阅连接器部分：https://nightlies.apache.org/flink/flink-docs-release-1.17/zh/docs/connectors/table/overview/
more >>

展开全文 >>

Flink Hudi DataStream API代码示例

前言

总结Flink通过DataStream API读写Hudi Demo示例，主要是自己备忘用。

  • 最开始学习Flink时都是使用Flink SQL,所以对于Flink SQL读写Hudi比较熟悉。但是对于写代码实现并不熟悉，而有些需求是基于Flink代码实现的，所以需要学习总结一下。
  • 仅为了实现用代码读写Hudi的需求，其实有两种方式，一种是在代码里通过Flink Table API，也就是代码中执行Flink SQL，这种方式其实和通过SQL实现差不多，另一种方式是通过DataStream API实现。（现实中包括网上教程使用最多的应该是Flink Table API）
  • 本文主要是总结DataStream API方式
  • DataStream API方式有一种好处是方便IDEA本地调试Hudi源码，便于学习，当然Table API也是可以进行本地调试源码的，但是因为我对Flink SQL源码不熟悉，调试起来比较费劲。Table API调试源码的难点在于我不知道从Flink SQL的源码到Hudi源码的入口在哪，因为这里牵扯到SQL解析的源码，可能比较麻烦（没有研究过）。比如我之前总结的Hudi Spark SQL源码相关的文章：Hudi Spark SQL源码学习总结-Create Table
more >>

展开全文 >>

Flink MySQL CDC 使用总结

前言

学习总结Flink MySQL CDC，主要目的是同步MySQL数据至其他数据源如Hudi、MySQL等，本文主要以 MySQL2Hudi、MySQL2MySQL两个场景进行示例验证。

版本

Flink 版本
Flink 1.14.3、1.15.4、1.16.1
Hudi 0.13.0
MYSQL CDC 2.3.0

安装

将下面的Jar包拷贝到flink/lib下面 （以flink1.15.4为例）

Flink CDC,只是对于Source表，比如MySQL CDC，就是抽取MySQL Source表，CDC 官方文档：https://ververica.github.io/flink-cdc-connectors/master/content/connectors/mysql-cdc.html#,可以查看官方文档了解目前Flink CDC支持哪些数据源，每一种数据源都需要下载对应的Jar包

more >>

展开全文 >>

Flink SQL Checkpoint 学习总结

前言

学习总结Flink SQL Checkpoint的使用，主要目的是为了验证Flink SQL流式任务挂掉后，重启时还可以继续从上次的运行状态恢复。

验证方式

Flink SQL流式增量读取Hudi表然后sink MySQL表，任务启动后处于running状态，先查看sink表有数据，然后将对应的yarn kill掉，再通过设置的checkpoint重启任务，任务重启后验证sink表的数据量。Flink SQL流式增量读取Hudi表可以参考：Flink SQL增量查询Hudi表

版本

  • Flink 1.14.3
  • Hudi 0.13.0

Checkpoint 参数

一般需要设置的常用参数

1
2
3
4
5
6
7
8
9
-- checkpoint间隔时间，单位毫秒，没有默认值，如果想开启checkpoint，需要将该参数设置一个大于0的数值
-- 如果想提升sink性能，比如写hudi，需要将该值设置大一点，因为间隔时间决定了批次大小
-- checkpoint间隔时间不能设置太短也不能设置太长，太短影响写入性能，太长影响数据及时性。
set execution.checkpointing.interval=1000;
-- 保存checkpoint文件的目录
set state.checkpoints.dir=hdfs:///flink/checkpoints/hudi2mysql;
-- 任务取消后保留checkpoint,默认值NO_EXTERNALIZED_CHECKPOINTS，
-- 可选值NO_EXTERNALIZED_CHECKPOINTS、DELETE_ON_CANCELLATION、RETAIN_ON_CANCELLATION
set execution.checkpointing.externalized-checkpoint-retention=RETAIN_ON_CANCELLATION;

从checkpoint恢复

1
set execution.savepoint.path=hdfs:///flink/checkpoints/hudi2mysql/255bdd01cee7486113feb1cbe8b45ee0/chk-1314;

其他参数

1
2
3
4
5
6
7
-- checkpoint模式，默认值EXACTLY_ONCE，可选值：EXACTLY_ONCE、AT_LEAST_ONCE
-- 要想支持EXACTLY_ONCE，需要sink端支持事务
set execution.checkpointing.mode=EXACTLY_ONCE;
-- checkpoint超时时间，默认10分钟
set execution.checkpointing.timeout=600000;
-- checkpoint文件保留数，默认1
set state.checkpoints.num-retained=3;
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>目前在济南一家公司从事大数据的工作~~