SqlServer数据库需支持基于flink-cdc实时读connector

## 模块功能说明
基于[Flink-CDC](https://github.com/apache/flink-cdc) 的 [SqlServer Source Connector](https://nightlies.apache.org/flink/flink-cdc-docs-release-3.2/docs/connectors/flink-sources/sqlserver-cdc/)组件，在TIS中为其封装成UI式组件，零代码构建SqlServer作为源，与多种目标端（例如：MySQL、Doris、ElasticSearch等，TIS支持的目标端类型可[点此链接](https://tis.pub/docs/#tis%E6%94%AF%E6%8C%81%E7%9A%84dataxreaderwriterplugin)）对接的实时数据通道。

下图是 **TIS SqlServer Source Connector** 组件相关的类图（后端）及利用渲染技术自动生成的前端部分页面效果：
![输入图片说明](https://foruda.gitee.com/images/1731290542952655936/9efe2da5_2303652.jpeg "实时增量框架类图.jpg") [详细大图](https://www.processon.com/view/link/673065a66b5a4a4addb9a8be?cid=646380819ea73956125b947b)

经过迭代TIS已经构建起一套构建数据集成生态的SDK包，普通开发者可以借助这套SDK**快速创建出新TIS前后端一体数据端**，正如上图，本次迭代开发**SqlServer Source Connector**组件，开发者只需要完成上图中两个**紫色的模块类**功能实现，TIS在框架会自动编译打包成TIS的专属插件包，便可在生产环境中部署，便为用户提供开箱即用的数据集成工具。

## 需求点列表及说明
### 列表

| 资源名  | 功能说明  |  参考 |
|---|---|---|
|  `FlinkCDCSqlServerSourceFactory.java` |  负责监听数据源增量日志所创建必要的构建的工厂类（类似监听MySql，PG中数据源所对应的Flink SourceFunction 封装），该类作为TIS与Flink-CDC SqlServer Source Connector 桥接器。 | [FlinkCDCMySQLSourceFactory.java]( https://gitee.com/qlangtech/plugins/blob/master/tis-incr/tis-flink-cdc-mysql-plugin/src/main/java/com/qlangtech/tis/plugins/incr/flink/cdc/mysql/FlinkCDCMySQLSourceFactory.java) |
|  `FlinkCDCSqlServerSourceFactory.json` `FlinkCDCSqlServerSourceFactory.md` | FlinkCDCSqlServerSourceFactory类内表单输入项说明清单，用于在UI表单中生成Help内容信息  |  [FlinkCDCMySQLSourceFactory.json](https://gitee.com/qlangtech/plugins/blob/master/tis-incr/tis-flink-cdc-mysql-plugin/src/main/resources/com/qlangtech/tis/plugins/incr/flink/cdc/mysql/FlinkCDCMySQLSourceFactory.json) [FlinkCDCMySQLSourceFactory.md](https://gitee.com/qlangtech/plugins/blob/master/tis-incr/tis-flink-cdc-mysql-plugin/src/main/resources/com/qlangtech/tis/plugins/incr/flink/cdc/mysql/FlinkCDCMySQLSourceFactory.md) |
| `FlinkCDCSqlServerSourceFunction.java`  | Flink实时任务实例启动入口  |  [FlinkCDCMysqlSourceFunction.java](https://gitee.com/qlangtech/plugins/blob/master/tis-incr/tis-flink-cdc-mysql-plugin/src/main/java/com/qlangtech/tis/plugins/incr/flink/cdc/mysql/FlinkCDCMysqlSourceFunction.java) |
| `TestFlinkCDCSqlServerSourceFactoryE2E.java`  | TestFlinkCDCSqlServerSourceFactory类对应的端到端单元测试类，利用 SqlServer Docker Container实现SqlServer的端到端集成测试|   |
| `TestFlinkCDCSqlServerSourceFactory.java`  | TestFlinkCDCSqlServerSourceFactory类对应的单元测试类，用于测试内部执行逻辑是否符合预期|   |
| `SqlServerCDCTypeVisitor.java`  |  对Flink CDC传入的DTO对象内的数据进行适配，封装成Flink内部的RowData数据结构 |  [MySQLCDCTypeVisitor.java](https://gitee.com/qlangtech/plugins/blob/master/tis-incr/tis-flink-cdc-mysql-plugin/src/main/java/com/qlangtech/tis/plugins/incr/flink/cdc/mysql/FlinkCDCMysqlSourceFunction.java#L117) |
|一篇MarkDown格式的公众号文章|记录本次众包任务开发迭代的总结，可以记录开发过程的点滴，所思、所想、对于TIS的建议或者意见，可以使TIS变得更好。届时会刊发在**TIS实验室** 微信公众号中||

以上清单中提到的资源文件已经在[脚手架项目](https://github.com/qlangtech/tis-sqlserver-plugin)中初始化，在项目中按名称查找即可。

### 详细说明
1. `FlinkCDCSqlServerSourceFactory.java`
    以下是该类的成员属性说明（TIS会自动将这些属性渲染到前端页面作为输入项）
    1. startupOptions
       * **initial** (default): Takes a snapshot of structure and data of captured tables; useful if topics should be populated with a complete representation of the data from the captured tables.（历史全量执行完后自动切换到最新变更数据）
       * **initial-only**: Takes a snapshot of structure and data like initial but instead does not transition into streaming changes once the snapshot has completed.(只做历史全量)
       * **latest-offset**: Takes a snapshot of the structure of captured tables only; useful if only changes happening from now onwards should be propagated to topics. （从数据当前最新变更开始监听）
      
       对应Flink-CDC的`org.apache.flink.cdc.connectors.base.options.StartupOptions`类
    2. timeZone 
       服务端所在时区，对应 [SqlServer Connector-Options](https://nightlies.apache.org/flink/flink-cdc-docs-release-3.2/docs/connectors/flink-sources/sqlserver-cdc/#connector-options)的 **server-time-zone** 属性
    3. independentBinLogMonitor 同库多表监听，没个表是否独占一个solt资源？
    
    以上属性实现可参考 [FlinkCDCMySQLSourceFactory.java](https://gitee.com/qlangtech/plugins/blob/master/tis-incr/tis-flink-cdc-mysql-plugin/src/main/java/com/qlangtech/tis/plugins/incr/flink/cdc/mysql/FlinkCDCMySQLSourceFactory.java)
2. `FlinkCDCSqlServerSourceFunction.java` 
    利用表单中的输入项值，构建SqlServer Flink-CDC Source增量数据监听器（**此类是本次开发的重点**）

## 脚手架
TIS 已经为本组件提供了脚手架项目，开发者只需要在此项目上继续开发完成`模块列表`中提及的模块。脚手架地址为 https://github.com/qlangtech/tis-sqlserver-plugin/tree/dev/tis-flink-cdc-sqlserver-plugin

项目中包含三个子Module：
1. **tis-flink-cdc-sqlserver-plugin**
   内实现了基于Flink-CDC SqlServer 的TIS 源端Connector组件
    ``` xml
    <dependency>
       <groupId>org.apache.flink</groupId>
       <artifactId>flink-connector-sqlserver-cdc</artifactId>
       
       <version>${flink.cdc.version}</version>
    </dependency>
    ```
2. **tis-flink-chunjun-sqlserver-plugin**
   内实现了基于chunjun的SqlServer的TIS目标端Connector组件（内部代码已经编写完成，并且单元测试已经跑通，开发者有兴趣可以借鉴一下）

3. **tis-testcontainer-sqlserver**
   封装了SqlServer testContainer组件用于E2E测试中用docker容器运行一个SqlServer数据端作为测试Stub（桩）用。tis-flink-cdc-sqlserver-plugin Module中可以将此Module作为Artifact依赖。具体使用方法可以借鉴`tis-flink-chunjun-sqlserver-plugin` Module中的例子。

## 开发流程
本模块开发流程分为以下几个步骤：
1. 模块编写
2. 单元测试编写
3. 集成测试
   1. 进入`tis-flink-cdc-sqlserver-plugin`模块，执行以下命令：
        ```shell
         mvn tis:run
        ```
       打开浏览器: `http://localhost:8080` 即可打开TIS，进行验证测试
   2. 下载安装flink执行包
      * 下载 TIS订制Flink执行包：http://mirror.qlangtech.com/4.1.0-SNAPSHOT/tis/flink-tis-1.18.1-bin.tar.gz
      * 修改配置文件 `FLINK_HOME/conf/flink-conf.yaml`
        添加JVM参数(修改**env.java.opts.all**配置项)`-Dskip_classloader_factory_creation=true -Ddata.dir=${USER_HOME}/tis_tmp/uber-data/data`，修改之后 内容如下：
        ```yaml
        env.java.opts.all: -Dskip_classloader_factory_creation=true -Ddata.dir=${USER_HOME}/tis_tmp/uber-data/data --add-exports=java.base/sun.net.util=ALL-UNNAMED --add-exports=java.rmi/sun.rmi.registry=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.api=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.file=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.parser=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.tree=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.util=ALL-UNNAMED --add-exports=java.security.jgss/sun.security.krb5=ALL-UNNAMED --add-opens=java.base/java.lang=ALL-UNNAMED --add-opens=java.base/java.net=ALL-UNNAMED --add-opens=java.base/java.io=ALL-UNNAMED --add-opens=java.base/java.nio=ALL-UNNAMED --add-opens=java.base/sun.nio.ch=ALL-UNNAMED --add-opens=java.base/java.lang.reflect=ALL-UNNAMED --add-opens=java.base/java.text=ALL-UNNAMED --add-opens=java.base/java.time=ALL-UNNAMED --add-opens=java.base/java.util=ALL-UNNAMED --add-opens=java.base/java.util.concurrent=ALL-UNNAMED --add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED --add-opens=java.base/java.util.concurrent.locks=ALL-UNNAMED
        ```
      * 启动Flink，使用浏览器打开flink地址：`http://localhost:8081`
        ``` shell 
        cd FLINK_HOME
        ./bin/start-cluster.sh
        ``` 
         
4. 提交代码，通知TIS团队同学验收
5. 撰写本次开发历程总结

## 开发环境

* 安装JDK11 环境
* Apache Maven 3.9.9 
* 本地github客户端可提交代码
* MacOS

## 开发模式
本项目使用众包开发模式，届时会在[oschina 众包平台](https://zb.oschina.net/) 发布众包任务，可由符合要求的开发者认领。
以下是TIS众包流程相关说明：
1. 如何参与：https://tis.pub/docs/co-construction/take-part-in/
2. 任务的生命周期：https://tis.pub/docs/co-construction/crowdsourcing-lifecycle/

## TIS介绍

![输入图片说明](https://gitee.com/qlangtech/tis/raw/master/docs/tis-logo.png "在这里输入图片标题")
TIS为您提供企业级数据集成服务，基于批(DataX)，流(Flink-CDC、Chunjun)一体，提供简单易用的操作界面，降低用户实施端到端数据同步的实施门槛，缩短任务配置时间，避免配置过程中出错，使数据同步变得简单、有趣且容易上手 [详细介绍](https://tis.pub/docs/)

有意向接包的同学如果初次听说TIS，尚不了解，可以先从官网上下载TIS Release安装包在本地试用。或者观看以下TIS操作视频：

1. [MySQL导入ElasticSearch](https://www.bilibili.com/video/BV1G64y1B7wm/?share_source=copy_web)
2. [MySQL导入Hive](https://www.bilibili.com/video/BV1Vb4y1z7DN/?share_source=copy_web)
3. [MySQL导入Clickhouse](https://www.bilibili.com/video/BV1x64y1B7V8/)
4. [MySQL到StarRocks的同步实现](https://tis.pub/docs/example/mysql-sync-starrocks) >>[视频Demo](https://www.bilibili.com/video/BV1Ah411x7J8/?zw&vd_source=fb348e27d4839f46057045d150b06dce)
5. MySQL同步Doris [>批量同步](https://www.bilibili.com/video/BV1eh4y1o7yQ/) [>实时同步](https://www.bilibili.com/video/BV1nX4y1h7SW/)

## 资料参考
1. flinkCDC 支持 sqlserver source connector，https://nightlies.apache.org/flink/flink-cdc-docs-release-3.2/docs/connectors/flink-sources/sqlserver-cdc/#single-thread-reading
2. [TIS Flink-CDC MySQL 封装实现](https://github.com/qlangtech/plugins/blob/master/tis-incr/tis-flink-cdc-mysql-plugin) 可作为SqlServer Connector实现的参考，两个Source Connector由于都是基于Flink-CDC实现的，因此内部类结构基本一致。 
3. [TIS实时增量框架类图](https://www.processon.com/view/link/673065a66b5a4a4addb9a8be?cid=646380819ea73956125b947b) [编辑](https://www.processon.com/diagraming/646380819ea73956125b9478)

qlangtech/tis

内容风险标识

评论 (0)

qlangtech/tis .gitee-modal { width: 500px !important; }

内容风险标识