之前在Flink Hudi DataStream API代码示例中总结了Hudi Flink DataStream API的代码及本地调试，并且在文中提到其实大家用Table API更多一些，但是我感觉Table API调试源码可能会比较难一点，因为可能会涉及到SQL解析，不清楚Table API的入口在哪里。
但是在我总结的上篇文章Flink用户自定义连接器（Table API Connectors）学习总结中知道了其实Flink Table API读写Hudi是通过自定义实现了一个Hudi的Table API Connectors（’connector’ = ‘hudi’）,相关类为HoodieTableFactory、HoodieTableSink和HoodieTableSource，入口为HoodieTableFactory。
结合官网文档和自定义实现一个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通过DataStream API读写Hudi Demo示例，主要是自己备忘用。
学习总结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为例）
more >>Flink CDC,只是对于Source表，比如MySQL CDC，就是抽取MySQL Source表，CDC 官方文档：https://ververica.github.io/flink-cdc-connectors/master/content/connectors/mysql-cdc.html#,可以查看官方文档了解目前Flink CDC支持哪些数据源，每一种数据源都需要下载对应的Jar包
学习总结Flink SQL Checkpoint的使用，主要目的是为了验证Flink SQL流式任务挂掉后，重启时还可以继续从上次的运行状态恢复。
Flink SQL流式增量读取Hudi表然后sink MySQL表，任务启动后处于running状态，先查看sink表有数据，然后将对应的yarn kill掉，再通过设置的checkpoint重启任务，任务重启后验证sink表的数据量。Flink SQL流式增量读取Hudi表可以参考：Flink SQL增量查询Hudi表
一般需要设置的常用参数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;
1 | set execution.savepoint.path=hdfs:///flink/checkpoints/hudi2mysql/255bdd01cee7486113feb1cbe8b45ee0/chk-1314; |
1 | -- checkpoint模式，默认值EXACTLY_ONCE，可选值：EXACTLY_ONCE、AT_LEAST_ONCE |
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