FlinkCDC全量及增量采集SqlServer数据

玛卡巴卡的卡巴卡玛  金牌会员 | 2024-8-3 09:04:51 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 921|帖子 921|积分 2763

​
   本文将具体先容Flink-CDC如何全量及增量采集Sqlserver数据源，准备适配Sqlserver数据源的小伙伴们可以参考本文，盼望本文能给你带来肯定的资助。
  一、Sqlserver的安装及开启事务日志

如果没有Sqlserver情况，但你又想学习这块的内容，那你只能自己动手通过docker安装一个 myself sqlserver来用作学习，当然，如果你有现成情况，那就查抄一下Sqlserver是否开启了代理(sqlagent.enabled)服务和CDC功能。
1.1 docker拉取镜像

看Github上写Flink-CDC 现在支持的Sqlserver版本为2012, 2014, 2016, 2017, 2019，但我想全部拉到最新（究竟证明，2022-latest 和latest是一样的，由于imagId都是一致的，且在后续测试也是没有问题的），以是我在docker上拉取镜像时，直接采用如下命令：
  1. docker pull mcr.microsoft.com/mssql/server:latest
复制代码
1.2 运行Sqlserver并设置代理

标准启动模式，没什么好说的，主要设置一下暗码(暗码要求比力严格，建议直接在网上搜个随机暗码生成器来搞一下)。
  1. docker run -e 'ACCEPT_EULA=Y' -e 'SA_PASSWORD=${your_password}' \
  2.    -p 1433:1433 --name sqlserver \
  3.    -d mcr.microsoft.com/mssql/server:latest
复制代码
设置代理sqlagent.enabled,代理设置完成后，需要重启Sqlserver，由于我们是docker安装的，直接用docker restart sqlserver就行了。
  1. [root@hdp-01 ~]# docker exec -it --user root sqlserver bash
  2. root@0274812d0c10:/# /opt/mssql/bin/mssql-conf set sqlagent.enabled true
  3. SQL Server needs to be restarted in order to apply this setting. Please run
  4. 'systemctl restart mssql-server.service'.
  5. root@0274812d0c10:/# exit
  6. exit
  7. [root@hdp-01 ~]# docker restart sqlserver
  8. sqlserver
复制代码
1.3 启用CDC功能

按照如下步调执行命令，如果看到is_cdc_enabled = 1，则说明当前数据库
  1. root@0274812d0c10:/# /opt/mssql-tools/bin/sqlcmd -S localhost -U SA -P "${your_password}"
  2. 1> create databases test;
  3. 2> go
  4. 1> use test;
  5. 2> go
  6. Changed database context to 'test'.
  7. 1> EXEC sys.sp_cdc_enable_db;
  8. 2> go
  9. 1> SELECT is_cdc_enabled FROM sys.databases WHERE name = 'test';
  10. 2> go
  11. is_cdc_enabled
  12. --------------
  13.              1
  15. (1 rows affected)
  16. 1> CREATE TABLE t_info (id int,order_date date,purchaser int,quantity int,product_id int,PRIMARY KEY ([id]))
  17. 2> go
  18. 1>
  19. 2>
  20. 3> EXEC sys.sp_cdc_enable_table
  21. 4> @source_schema = 'dbo',
  22. 5> @source_name   = 't_info',
  23. 6> @role_name     = 'cdc_role';
  24. 7> go
  25. Update mask evaluation will be disabled in net_changes_function because the CLR configuration option is disabled.
  26. Job 'cdc.zeus_capture' started successfully.
  27. Job 'cdc.zeus_cleanup' started successfully.
  28. 1> select * from t_info;
  29. 2> go
  30. id          order_date       purchaser   quantity    product_id
  31. ----------- ---------------- ----------- ----------- -----------
  33. (0 rows affected)
复制代码
1.4 查抄CDC是否正常开启

用客户端连接Sqlserver，检察test库下的INFORMATION_SCHEMA.TABLES中是否出现TABLE_SCHEMA = cdc的表，如果出现，说明已经乐成安装Sqlserver并启用了CDC。
  1. 1> use test;
  2. 2> go
  3. Changed database context to 'test'.
  4. 1> select * from INFORMATION_SCHEMA.TABLES;
  5. 2> go
复制代码
  1. TABLE_CATALOG        TABLE_SCHEMA        TABLE_NAME               TABLE_TYPE
  2. test                    dbo              user_info                 BASE TABLE
  3. test                    dbo              systranschemas           BASE TABLE
  4. test                    cdc              change_tables             BASE TABLE
  5. test                    cdc              ddl_history               BASE TABLE
  6. test                    cdc              lsn_time_mapping         BASE TABLE
  7. test                    cdc              captured_columns         BASE TABLE
  8. test                    cdc              index_columns             BASE TABLE
  9. test                    dbo              orders                   BASE TABLE
  10. test                    cdc              dbo_orders_CT             BASE TABLE
复制代码
二、具体实现

2.1 Flik-CDC采集SqlServer主程序

添加依靠包：
  1.         <dependency>
  2.             <groupId>com.ververica</groupId>
  3.             <artifactId>flink-connector-sqlserver-cdc</artifactId>
  4.             <version>3.0.0</version>
  5.         </dependency>
复制代码
编写主函数：
  1.     public static void main(String[] args) throws Exception {
  3.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  5.         // 设置全局并行度
  6.         env.setParallelism(1);
  7.         // 设置时间语义为ProcessingTime
  8.         env.getConfig().setAutoWatermarkInterval(0);
  9.         // 每隔60s启动一个检查点
  10.         env.enableCheckpointing(60000, CheckpointingMode.EXACTLY_ONCE);
  11.         // checkpoint最小间隔
  12.         env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1000);
  13.         // checkpoint超时时间
  14.         env.getCheckpointConfig().setCheckpointTimeout(60000);
  15.         // 同一时间只允许一个checkpoint
  16.         // env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
  17.         // Flink处理程序被cancel后，会保留Checkpoint数据
  18.         //   env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
  21.         SourceFunction<String> sqlServerSource = SqlServerSource.<String>builder()
  22.                 .hostname("localhost")
  23.                 .port(1433)
  24.                 .username("SA")
  25.                 .password("")
  26.                 .database("test")
  27.                 .tableList("dbo.t_info")
  28.                 .startupOptions(StartupOptions.initial())
  29.                 .debeziumProperties(getDebeziumProperties())
  30.                 .deserializer(new CustomerDeserializationSchemaSqlserver())
  31.                 .build();
  33.         DataStreamSource<String> dataStreamSource = env.addSource(sqlServerSource, "_transaction_log_source");
  34.         dataStreamSource.print().setParallelism(1);
  35.         env.execute("sqlserver-cdc-test");
  37.     }
  38.    
  39.    
  40.         public static Properties getDebeziumProperties() {
  41.         Properties properties = new Properties();
  42.         properties.put("converters", "sqlserverDebeziumConverter");
  43.         properties.put("sqlserverDebeziumConverter.type", "SqlserverDebeziumConverter");
  44.         properties.put("sqlserverDebeziumConverter.database.type", "sqlserver");
  45.         // 自定义格式，可选
  46.         properties.put("sqlserverDebeziumConverter.format.datetime", "yyyy-MM-dd HH:mm:ss");
  47.         properties.put("sqlserverDebeziumConverter.format.date", "yyyy-MM-dd");
  48.         properties.put("sqlserverDebeziumConverter.format.time", "HH:mm:ss");
  49.         return properties;
  50.     }
复制代码
2.2 自界说Sqlserver反序列化格式：

Flink-CDC底层技能为debezium，它捕捉到Sqlserver数据变更(CRUD)的数据格式如下：
  1. #初始化
  2. Struct{after=Struct{id=1,order_date=2024-01-30,purchaser=1,quantity=100,product_id=1},source=Struct{version=1.9.7.Final,connector=sqlserver,name=sqlserver_transaction_log_source,ts_ms=1706574924473,snapshot=true,db=zeus,schema=dbo,table=orders,commit_lsn=0000002b:00002280:0003},op=r,ts_ms=1706603724432}
  4. #新增
  5. Struct{after=Struct{id=12,order_date=2024-01-11,purchaser=6,quantity=233,product_id=63},source=Struct{version=1.9.7.Final,connector=sqlserver,name=sqlserver_transaction_log_source,ts_ms=1706603786187,db=zeus,schema=dbo,table=orders,change_lsn=0000002b:00002480:0002,commit_lsn=0000002b:00002480:0003,event_serial_no=1},op=c,ts_ms=1706603788461}
  8. #更新
  9. Struct{before=Struct{id=12,order_date=2024-01-11,purchaser=6,quantity=233,product_id=63},after=Struct{id=12,order_date=2024-01-11,purchaser=8,quantity=233,product_id=63},source=Struct{version=1.9.7.Final,connector=sqlserver,name=sqlserver_transaction_log_source,ts_ms=1706603845603,db=zeus,schema=dbo,table=orders,change_lsn=0000002b:00002500:0002,commit_lsn=0000002b:00002500:0003,event_serial_no=2},op=u,ts_ms=1706603850134}
  12. #删除
  13. Struct{before=Struct{id=11,order_date=2024-01-11,purchaser=6,quantity=233,product_id=63},source=Struct{version=1.9.7.Final,connector=sqlserver,name=sqlserver_transaction_log_source,ts_ms=1706603973023,db=zeus,schema=dbo,table=orders,change_lsn=0000002b:000025e8:0002,commit_lsn=0000002b:000025e8:0005,event_serial_no=1},op=d,ts_ms=1706603973859}
复制代码
因此，可以根据自己需要自界说反序列化格式，将数据按照标准统一数据输出，下面是我自界说的格式，供各人参考：
  1. import com.alibaba.fastjson2.JSON;
  2. import com.alibaba.fastjson2.JSONObject;
  3. import com.alibaba.fastjson2.JSONWriter;
  4. import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
  5. import io.debezium.data.Envelope;
  6. import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
  7. import org.apache.flink.api.common.typeinfo.TypeInformation;
  8. import org.apache.flink.util.Collector;
  9. import org.apache.kafka.connect.data.Field;
  10. import org.apache.kafka.connect.data.Schema;
  11. import org.apache.kafka.connect.data.Struct;
  12. import org.apache.kafka.connect.source.SourceRecord;
  14. import java.util.HashMap;
  15. import java.util.Map;
  17. public class CustomerDeserializationSchemaSqlserver implements DebeziumDeserializationSchema<String> {
  19.     private static final long serialVersionUID = -1L;
  22.     @Override
  23.     public void deserialize(SourceRecord sourceRecord, Collector collector) {
  24.         Map<String, Object> resultMap = new HashMap<>();
  25.         String topic = sourceRecord.topic();
  26.         String[] split = topic.split("[.]");
  27.         String database = split[1];
  28.         String table = split[2];
  29.         resultMap.put("db", database);
  30.         resultMap.put("tableName", table);
  31.         //获取操作类型
  32.         Envelope.Operation operation = Envelope.operationFor(sourceRecord);
  33.         //获取数据本身
  34.         Struct struct = (Struct) sourceRecord.value();
  35.         Struct after = struct.getStruct("after");
  36.         Struct before = struct.getStruct("before");
  37.         String op = operation.name();
  38.         resultMap.put("op", op);
  40.         //新增,更新或者初始化
  41.         if (op.equals(Envelope.Operation.CREATE.name()) || op.equals(Envelope.Operation.READ.name()) || op.equals(Envelope.Operation.UPDATE.name())) {
  42.             JSONObject afterJson = new JSONObject();
  43.             if (after != null) {
  44.                 Schema schema = after.schema();
  45.                 for (Field field : schema.fields()) {
  46.                     afterJson.put(field.name(), after.get(field.name()));
  47.                 }
  48.                 resultMap.put("after", afterJson);
  49.             }
  50.         }
  52.         if (op.equals(Envelope.Operation.DELETE.name())) {
  53.             JSONObject beforeJson = new JSONObject();
  54.             if (before != null) {
  55.                 Schema schema = before.schema();
  56.                 for (Field field : schema.fields()) {
  57.                     beforeJson.put(field.name(), before.get(field.name()));
  58.                 }
  59.                 resultMap.put("before", beforeJson);
  60.             }
  61.         }
  63.         collector.collect(JSON.toJSONString(resultMap, JSONWriter.Feature.FieldBased, JSONWriter.Feature.LargeObject));
  65.     }
  67.     @Override
  68.     public TypeInformation<String> getProducedType() {
  69.         return BasicTypeInfo.STRING_TYPE_INFO;
  70.     }
  72. }
复制代码
2.3 自界说日期格式转换器

debezium会将日期转为5位数字，日期时间转为13位的数字，因此我们需要根据Sqlserver的日期类型转换成标准的时期或者时间格式。Sqlserver的日期类型主要包含以下几种：
字段类型快照类型(jdbc type)cdc类型(jdbc type)DATEjava.sql.Date(91)java.sql.Date(91)TIMEjava.sql.Timestamp(92)java.sql.Time(92)DATETIMEjava.sql.Timestamp(93)java.sql.Timestamp(93)DATETIME2java.sql.Timestamp(93)java.sql.Timestamp(93)DATETIMEOFFSETmicrosoft.sql.DateTimeOffset(-155)microsoft.sql.DateTimeOffset(-155)SMALLDATETIMEjava.sql.Timestamp(93)java.sql.Timestamp(93)
  1. import io.debezium.spi.converter.CustomConverter;
  2. import io.debezium.spi.converter.RelationalColumn;
  3. import org.apache.kafka.connect.data.SchemaBuilder;
  4. import java.time.ZoneOffset;
  5. import java.time.format.DateTimeFormatter;
  6. import java.util.Properties;
  8. @Sl4j
  9. public class SqlserverDebeziumConverter implements CustomConverter<SchemaBuilder, RelationalColumn> {
  13.     private static final String DATE_FORMAT = "yyyy-MM-dd";
  14.     private static final String TIME_FORMAT = "HH:mm:ss";
  15.     private static final String DATETIME_FORMAT = "yyyy-MM-dd HH:mm:ss";
  16.     private DateTimeFormatter dateFormatter;
  17.     private DateTimeFormatter timeFormatter;
  18.     private DateTimeFormatter datetimeFormatter;
  19.     private SchemaBuilder schemaBuilder;
  20.     private String databaseType;
  21.     private String schemaNamePrefix;
  24.     @Override
  25.     public void configure(Properties properties) {
  26.         // 必填参数：database.type，只支持sqlserver
  27.         this.databaseType = properties.getProperty("database.type");
  28.         // 如果未设置，或者设置的不是mysql、sqlserver，则抛出异常。
  29.         if (this.databaseType == null || !this.databaseType.equals("sqlserver"))) {
  30.             throw new IllegalArgumentException("database.type 必须设置为'sqlserver'");
  31.         }
  32.         // 选填参数：format.date、format.time、format.datetime。获取时间格式化的格式
  33.         String dateFormat = properties.getProperty("format.date", DATE_FORMAT);
  34.         String timeFormat = properties.getProperty("format.time", TIME_FORMAT);
  35.         String datetimeFormat = properties.getProperty("format.datetime", DATETIME_FORMAT);
  36.         // 获取自身类的包名+数据库类型为默认schema.name
  37.         String className = this.getClass().getName();
  38.         // 查看是否设置schema.name.prefix
  39.         this.schemaNamePrefix = properties.getProperty("schema.name.prefix", className + "." + this.databaseType);
  40.         // 初始化时间格式化器
  41.         dateFormatter = DateTimeFormatter.ofPattern(dateFormat);
  42.         timeFormatter = DateTimeFormatter.ofPattern(timeFormat);
  43.         datetimeFormatter = DateTimeFormatter.ofPattern(datetimeFormat);
  45.     }
  47.     // sqlserver的转换器
  48.     public void registerSqlserverConverter(String columnType, ConverterRegistration<SchemaBuilder> converterRegistration) {
  49.         String schemaName = this.schemaNamePrefix + "." + columnType.toLowerCase();
  50.         schemaBuilder = SchemaBuilder.string().name(schemaName);
  51.         switch (columnType) {
  52.             case "DATE":
  53.                 converterRegistration.register(schemaBuilder, value -> {
  54.                     if (value == null) {
  55.                         return null;
  56.                     } else if (value instanceof java.sql.Date) {
  57.                         return dateFormatter.format(((java.sql.Date) value).toLocalDate());
  58.                     } else {
  59.                         return this.failConvert(value, schemaName);
  60.                     }
  61.                 });
  62.                 break;
  63.             case "TIME":
  64.                 converterRegistration.register(schemaBuilder, value -> {
  65.                     if (value == null) {
  66.                         return null;
  67.                     } else if (value instanceof java.sql.Time) {
  68.                         return timeFormatter.format(((java.sql.Time) value).toLocalTime());
  69.                     } else if (value instanceof java.sql.Timestamp) {
  70.                         return timeFormatter.format(((java.sql.Timestamp) value).toLocalDateTime().toLocalTime());
  71.                     } else {
  72.                         return this.failConvert(value, schemaName);
  73.                     }
  74.                 });
  75.                 break;
  76.             case "DATETIME":
  77.             case "DATETIME2":
  78.             case "SMALLDATETIME":
  79.             case "DATETIMEOFFSET":
  80.                 converterRegistration.register(schemaBuilder, value -> {
  81.                     if (value == null) {
  82.                         return null;
  83.                     } else if (value instanceof java.sql.Timestamp) {
  84.                         return datetimeFormatter.format(((java.sql.Timestamp) value).toLocalDateTime());
  85.                     } else if (value instanceof microsoft.sql.DateTimeOffset) {
  86.                         microsoft.sql.DateTimeOffset dateTimeOffset = (microsoft.sql.DateTimeOffset) value;
  87.                         return datetimeFormatter.format(
  88.                                 dateTimeOffset.getOffsetDateTime().withOffsetSameInstant(ZoneOffset.UTC).toLocalDateTime());
  89.                     } else {
  90.                         return this.failConvert(value, schemaName);
  91.                     }
  92.                 });
  93.                 break;
  94.             default:
  95.                 schemaBuilder = null;
  96.                 break;
  97.         }
  98.     }
  101.     @Override
  102.     public void converterFor(RelationalColumn relationalColumn, ConverterRegistration<SchemaBuilder> converterRegistration) {
  103.         // 获取字段类型
  104.         String columnType = relationalColumn.typeName().toUpperCase();
  105.         // 根据数据库类型调用不同的转换器
  106.         if (this.databaseType.equals("sqlserver")) {
  107.             this.registerSqlserverConverter(columnType, converterRegistration);
  108.         } else {
  109.             log.warn("不支持的数据库类型: {}", this.databaseType);
  110.             schemaBuilder = null;
  111.         }
  112.     }
  114.     private String getClassName(Object value) {
  115.         if (value == null) {
  116.             return null;
  117.         }
  118.         return value.getClass().getName();
  119.     }
  121.     // 类型转换失败时的日志打印
  122.     private String failConvert(Object value, String type) {
  123.         String valueClass = this.getClassName(value);
  124.         String valueString = valueClass == null ? null : value.toString();
  125.         return valueString;
  126.     }
  127. }
复制代码
三、总计

现在Fink-CDC对这种增量采集传统数据库的技能已经封装的很好了，并且官方也给了具体的操作教程，但如果想要深入的学习一项技能，个人觉得照旧要重新到尾操作一遍，一方面可以或许快速的提升自己，另一方面发现问题时，也能从不同的角度来思考办理方案，盼望本篇文章可以或许给各人带来一点资助。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

玛卡巴卡的卡巴卡玛

金牌会员
这个人很懒什么都没写!

楼主热帖

标签云

存储 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表