devbox
weixin_40725706
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Flink CDC整库同步_flinkcdc整库同步

作者：weixin_40725706 | 2024-08-23 06:44:00

赞

踩

flinkcdc整库同步
背景

项目需要能够捕获外部数据源的数据变更，实时同步到目标数据库中，自动更新数据，实现源数据库和目标数据库所有表的数据同步更新，本文以mysql -> greenplumn场景记录实现方案。

实现
1.引入依赖
  1. <dependency>
  2.     <groupId>com.ververica</groupId>
  3.     <artifactId>flink-sql-connector-mysql-cdc</artifactId>
  4.     <version>2.4.1</version>
  5. </dependency>
  6.  
  7. <!--        mysql-->
  8. <dependency>
  9.     <groupId>mysql</groupId>
  10.     <artifactId>mysql-connector-java</artifactId>
  11.     <version>8.0.29</version>
  12. </dependency>
  13. <dependency>
  14.     <groupId>org.postgresql</groupId>
  15.     <artifactId>postgresql</artifactId>
  16.     <version>42.5.0</version>
  17. </dependency>
2.创建FlinkCDCSource

创建FlinkCDC连接器，设置数据源的连接信息，日志捕获的起始时间点，读取并缓存数据源的元数据信息

  1. import com.ververica.cdc.connectors.mysql.source.MySqlSource;
  2. import com.ververica.cdc.connectors.mysql.table.StartupOptions;
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.apache.commons.lang3.StringUtils;
  5. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  6. import org.apache.flink.api.java.tuple.Tuple2;
  7. import org.apache.flink.connector.jdbc.catalog.MySqlCatalog;
  8. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  9. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  10. import org.apache.flink.table.api.Schema;
  11. import org.apache.flink.table.catalog.DefaultCatalogTable;
  12. import org.apache.flink.table.catalog.ObjectPath;
  13. import org.apache.flink.table.catalog.exceptions.TableNotExistException;
  14. import org.apache.flink.table.types.DataType;
  15. import org.apache.flink.table.types.logical.LogicalType;
  16. import org.apache.flink.table.types.logical.RowType;
  17. import org.apache.flink.types.Row;
  18.  
  19. import java.util.*;
  20. import java.util.stream.Collectors;
  21.  
  22. /**
  23.  * 目标表是否自动创建
  24.  * 表结构变更
  25.  */
  26. @Slf4j
  27. public class FlinkCdcMultiSyncJdbc {
  28.  
  29.     public static CdcPro cdcPro;
  30.  
  31.     public static MySqlCatalog mysqlCatalog;
  32.  
  33.     public static List<String> tables;
  34.  
  35.     public static void main(String[] args) throws Exception {
  36.  
  37.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  38.  
  39.         init();
  40.  
  41.         // 监控mysql binlog
  42.         //创建 Flink-MySQL-CDC 的 Source
  43.         //initial (default): 在第一次启动时对被监视的数据库表执行初始快照，并继续读取最新的binlog (开启断点续传后从上次消费offset继续消费)
  44.         //latest-offset: 永远不要在第一次启动时对被监视的数据库表执行快照，只从binlog的末尾读取，这意味着只有自连接器启动以来的更改
  45.         //timestamp: 永远不要在第一次启动时对监视的数据库表执行快照，直接从指定的时间戳读取binlog。使用者将从头遍历binlog，并忽略时间戳小于指定时间戳的更改事件
  46.         //specific-offset: 不允许在第一次启动时对监视的数据库表进行快照，直接从指定的偏移量读取binlog。
  47.         Properties jdbcProperties = new Properties();
  48.         jdbcProperties.setProperty("time_zone", "+8:00");
  49.         jdbcProperties.setProperty("serverTimeZone", "Asia/Shanghai");
  50.         MySqlSource<Tuple2<String, Row>> mySqlSource = MySqlSource.<Tuple2<String, Row>>builder()
  51.                 .hostname(cdcPro.getHost())
  52.                 .port(cdcPro.getPort())
  53.                 .databaseList(cdcPro.getDb())
  54.                 .tableList(tables.toArray(new String[0]))
  55. //                .scanNewlyAddedTableEnabled(true) // 启用扫描新添加的表功能
  56.                 .username(cdcPro.getUserName())
  57.                 .password(cdcPro.getPassWord())
  58.                 .jdbcProperties(jdbcProperties)
  59.                 .deserializer(new MySQLDebeziumDeserializer(buildTableRowTypeMap(cdcPro, tables)))
  60.                 .startupOptions(StartupOptions.latest())
  61.                 .includeSchemaChanges(false)
  62.                 .build();
  63.  
  64.         SingleOutputStreamOperator<Tuple2<String, Row>> dataStreamSource = env.fromSource(mySqlSource,
  65.                 WatermarkStrategy.noWatermarks(), "mysql cdc").disableChaining();
  66.         dataStreamSource.addSink(new DwSink(cdcPro));
  67.         env.execute("flinkcdc_mysql_dw");
  68.     }
  69.  
  70.     private static void init() throws Exception {
  71.         cdcPro = SourceUtil.createCdcPro();
  72.         String source_url = String.format("jdbc:mysql://%s:%d", cdcPro.getHost(), cdcPro.getPort());
  73.         // 注册同步的库对应的catalog
  74.         mysqlCatalog = new MySqlCatalog(new ClassLoader() {
  75.             @Override
  76.             public Class<?> loadClass(String name) throws ClassNotFoundException {
  77.                 return super.loadClass(name);
  78.             }
  79.         }, "mysql-catalog", cdcPro.getDb(), cdcPro.getUserName(), cdcPro.getPassWord(), source_url);
  80.         //        List<String> dbs = mysqlCatalog.listDatabases();
  81.         tables = buildTables(cdcPro);
  82.         System.setProperty("HADOOP_USER_NAME", "root");
  83.     }
  84.  
  85.     private static List<String> buildTables(CdcPro cdcPro) throws Exception {
  86.         // 如果整库同步，则从catalog里取所有表，否则从指定表中取表名
  87.         List<String> tables = new ArrayList<>();
  88.         if (".*".equals(cdcPro.getTableList())) {
  89.             tables = mysqlCatalog.listTables(cdcPro.getDb());
  90.         } else {
  91.             String[] tableArray = cdcPro.getTableList().split(",");
  92.             for (String table : tableArray) {
  93.                 tables.add(table);
  94.             }
  95.         }
  96.  
  97.         //过滤提出列表中的表清单
  98.         tables = tables.stream().filter(tableName->!cdcPro.getExcludeTables().contains(tableName)).map(t->cdcPro.getDb() + "." + t).collect(Collectors.toList());
  99.         log.info("cdc tables: \n" + tables);
  100.         return tables;
  101.     }
  102.  
  103.     private static Map<String, RowType> buildTableRowTypeMap(CdcPro cdcPro, List<String> tables) throws Exception {
  104.         Map<String, RowType> tableRowTypeMap = new HashMap<>();
  105. //        Map<String, RowTypeInfo> tableTypeInformationMap = Maps.newConcurrentMap();
  106.         tables.parallelStream().forEach(table ->{
  107.             String tableName = StringUtils.replace(table, cdcPro.getDb() + ".", "");
  108.             // 获取mysql catalog中注册的表
  109.             ObjectPath objectPath = new ObjectPath(cdcPro.getDb(), tableName);
  110.             DefaultCatalogTable catalogBaseTable = null;
  111.             try {
  112.                 catalogBaseTable = (DefaultCatalogTable) mysqlCatalog.getTable(objectPath);
  113.             } catch (TableNotExistException e) {
  114.                 throw new RuntimeException(e);
  115.             }
  116.             // 获取表的Schema
  117.             Schema schema = catalogBaseTable.getUnresolvedSchema();
  118.             // 获取表中字段名列表和表字段类型
  119.             String[] fieldNames = new String[schema.getColumns().size()];
  120.             LogicalType[] logicalTypes = new LogicalType[schema.getColumns().size()];
  121.             for (int i = 0; i < schema.getColumns().size(); i++) {
  122.                 Schema.UnresolvedPhysicalColumn column = (Schema.UnresolvedPhysicalColumn) schema.getColumns().get(i);
  123.                 fieldNames[i] = column.getName();
  124.                 logicalTypes[i] = ((DataType) column.getDataType()).getLogicalType();
  125.  
  126.             }
  127.             tableRowTypeMap.put(tableName, RowType.of(logicalTypes, fieldNames));
  128.         });
  129.         return tableRowTypeMap;
  130.     }
  131.  
  132. }
3.创建数据解析器

创建自定义数据解析器，根据FlinkCDC捕获的日志内容，解析出变更的表、数据、操作类型。

自定义数据转换器，将日志解析数据转换成为Flink数据类型

设置Flink数据的Changelog，表和数据组成二元组，放入Flink收集器中。

  1. import com.ververica.cdc.connectors.shaded.org.apache.kafka.connect.data.Decimal;
  2. import com.ververica.cdc.connectors.shaded.org.apache.kafka.connect.data.Field;
  3. import com.ververica.cdc.connectors.shaded.org.apache.kafka.connect.data.Schema;
  4. import com.ververica.cdc.connectors.shaded.org.apache.kafka.connect.data.Struct;
  5. import com.ververica.cdc.connectors.shaded.org.apache.kafka.connect.source.SourceRecord;
  6. import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
  7. import com.ververica.cdc.debezium.table.DeserializationRuntimeConverter;
  8. import com.ververica.cdc.debezium.utils.TemporalConversions;
  9. import io.debezium.data.Envelope;
  10. import io.debezium.data.SpecialValueDecimal;
  11. import io.debezium.data.VariableScaleDecimal;
  12. import io.debezium.time.*;
  13. import lombok.extern.slf4j.Slf4j;
  14. import org.apache.flink.api.common.typeinfo.TypeHint;
  15. import org.apache.flink.api.common.typeinfo.TypeInformation;
  16. import org.apache.flink.api.java.tuple.Tuple2;
  17. import org.apache.flink.calcite.shaded.com.google.common.collect.Maps;
  18. import org.apache.flink.table.catalog.DefaultCatalogTable;
  19. import org.apache.flink.table.catalog.ObjectPath;
  20. import org.apache.flink.table.data.DecimalData;
  21. import org.apache.flink.table.data.StringData;
  22. import org.apache.flink.table.data.TimestampData;
  23. import org.apache.flink.table.types.DataType;
  24. import org.apache.flink.table.types.logical.DecimalType;
  25. import org.apache.flink.table.types.logical.LogicalType;
  26. import org.apache.flink.table.types.logical.RowType;
  27. import org.apache.flink.types.Row;
  28. import org.apache.flink.types.RowKind;
  29. import org.apache.flink.types.RowUtils;
  30. import org.apache.flink.util.Collector;
  31.  
  32. import java.math.BigDecimal;
  33. import java.nio.ByteBuffer;
  34. import java.time.Instant;
  35. import java.time.LocalDateTime;
  36. import java.time.LocalTime;
  37. import java.time.ZoneId;
  38. import java.util.LinkedHashMap;
  39. import java.util.Map;
  40. import java.util.Properties;
  41.  
  42. @Slf4j
  43. public class MySQLDebeziumDeserializer implements DebeziumDeserializationSchema<Tuple2<String, Row>> {
  44.  
  45.     private final Map<String, RowType> tableRowTypeMap;
  46.     private final Map<String, DeserializationRuntimeConverter> physicalConverterMap = Maps.newConcurrentMap();
  47.  
  48.     MySQLDebeziumDeserializer(Map<String, RowType> tableRowTypeMap) {
  49.         this.tableRowTypeMap = tableRowTypeMap;
  50.         for (String tableName : this.tableRowTypeMap.keySet()) {
  51.             RowType rowType = this.tableRowTypeMap.get(tableName);
  52.             DeserializationRuntimeConverter physicalConverter = createNotNullConverter(rowType);
  53.             this.physicalConverterMap.put(tableName, physicalConverter);
  54.         }
  55.     }
  56.  
  57.     @Override
  58.     public void deserialize(SourceRecord record, Collector out) throws Exception {
  59.         Struct value = (Struct) record.value();
  60. //        log.info("value:{}", value);
  61.         String tableName = value.getStruct("source").get("table").toString();
  62.  
  63.         //如果该表不在初始化的列表中，过滤该表的操作
  64.         if(!physicalConverterMap.containsKey(tableName)){
  65.             return;
  66.         }
  67.  
  68.         DeserializationRuntimeConverter physicalConverter = physicalConverterMap.get(tableName);
  69.         Envelope.Operation op = Envelope.operationFor(record);
  70.  
  71.         if (op == Envelope.Operation.CREATE || op == Envelope.Operation.READ) {
  72.             Row insert = extractAfterRow(record, physicalConverter);
  73.             insert.setKind(RowKind.INSERT);
  74.             out.collect(Tuple2.of(tableName, insert));
  75.         } else if (op == Envelope.Operation.DELETE) {
  76.             Row delete = extractBeforeRow(record, physicalConverter);
  77.             delete.setKind(RowKind.DELETE);
  78.             out.collect(Tuple2.of(tableName, delete));
  79.         } else if (op == Envelope.Operation.UPDATE) {
  80.             Row before = extractBeforeRow(record, physicalConverter);
  81.             before.setKind(RowKind.UPDATE_BEFORE);
  82.             out.collect(Tuple2.of(tableName, before));
  83.  
  84.             Row after = extractAfterRow(record, physicalConverter);
  85.             after.setKind(RowKind.UPDATE_AFTER);
  86.             out.collect(Tuple2.of(tableName, after));
  87.         }
  88.  
  89.     }
  90.  
  91.     private Row extractAfterRow(SourceRecord record, DeserializationRuntimeConverter physicalConverter) throws Exception {
  92.         Schema afterSchema = record.valueSchema().field(Envelope.FieldName.AFTER).schema();
  93.         Struct after = ((Struct) record.value()).getStruct(Envelope.FieldName.AFTER);
  94.         return (Row) physicalConverter.convert(after, afterSchema);
  95.     }
  96.  
  97.     private Row extractBeforeRow(SourceRecord record, DeserializationRuntimeConverter physicalConverter) throws Exception {
  98.         Schema beforeSchema = record.valueSchema().field(Envelope.FieldName.BEFORE).schema();
  99.         Struct before = ((Struct) record.value()).getStruct(Envelope.FieldName.BEFORE);
  100.         return (Row) physicalConverter.convert(before, beforeSchema);
  101.     }
  102.  
  103.     @Override
  104.     public TypeInformation<Tuple2<String, Row>> getProducedType() {
  105.         return TypeInformation.of(new TypeHint<Tuple2<String, Row>>() {
  106.         });
  107.     }
  108.  
  109.     public static DeserializationRuntimeConverter createNotNullConverter(LogicalType type) {
  110.         /**
  111.          * datetime       TIMESTAMP_WITHOUT_TIME_ZONE
  112.          * date           DATE
  113.          * time           TIME_WITHOUT_TIME_ZONE
  114.          * timestamp      TIMESTAMP_WITHOUT_TIME_ZONE
  115.          */
  116.         switch (type.getTypeRoot()) {
  117.             case NULL:
  118.                 return convertToNull();
  119.             case BOOLEAN:
  120.                 return convertToInt();
  121.             case TINYINT:
  122.                 return convertToInt();
  123.             case SMALLINT:
  124.                 return convertToInt();
  125.             case INTEGER:
  126.             case INTERVAL_YEAR_MONTH:
  127.                 return convertToInt();
  128.             case BIGINT:
  129.             case INTERVAL_DAY_TIME:
  130.                 return convertToLong();
  131.             case DATE:
  132.                 return convertToDate();
  133.             case TIME_WITHOUT_TIME_ZONE:
  134.                 return convertToTime();
  135.             case TIMESTAMP_WITHOUT_TIME_ZONE:
  136.                 return convertToTimestamp(ZoneId.of("UTC+8"));
  137.             case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
  138.                 return convertToLocalTimeZoneTimestamp(ZoneId.of("UTC"));
  139.             case FLOAT:
  140.                 return convertToFloat();
  141.             case DOUBLE:
  142.                 return convertToDouble();
  143.             case CHAR:
  144.             case VARCHAR:
  145.                 return convertToString();
  146.             case BINARY:
  147.             case VARBINARY:
  148.                 return convertToBinary();
  149.             case DECIMAL:
  150.                 return createDecimalConverter((DecimalType) type);
  151.             case ROW:
  152.                 return createRowConverter((RowType) type);
  153.             case ARRAY:
  154.             case MAP:
  155.             case MULTISET:
  156.             case RAW:
  157.             default:
  158.                 throw new UnsupportedOperationException("Unsupported type: " + type);
  159.         }
  160.     }
  161.  
  162.     private static DeserializationRuntimeConverter convertToNull() {
  163.         return new DeserializationRuntimeConverter() {
  164.  
  165.             private static final long serialVersionUID = 1L;
  166.  
  167.             @Override
  168.             public Object convert(Object dbzObj, Schema schema) {
  169.                 return null;
  170.             }
  171.  
  172.         };
  173.     }
  174.  
  175.     private static DeserializationRuntimeConverter convertToByte() {
  176.         return new DeserializationRuntimeConverter() {
  177.  
  178.             private static final long serialVersionUID = 1L;
  179.  
  180.             @Override
  181.             public Object convert(Object dbzObj, Schema schema) {
  182.                 return Byte.parseByte(dbzObj.toString());
  183.             }
  184.  
  185.         };
  186.     }
  187.  
  188.     private static DeserializationRuntimeConverter convertToBoolean() {
  189.         return new DeserializationRuntimeConverter() {
  190.             private static final long serialVersionUID = 1L;
  191.  
  192.             @Override
  193.             public Object convert(Object dbzObj, Schema schema) {
  194.                 if (dbzObj instanceof Boolean) {
  195.                     return dbzObj;
  196.                 } else if (dbzObj instanceof Byte) {
  197.                     return (byte) dbzObj == 1;
  198.                 } else if (dbzObj instanceof Short) {
  199.                     return (short) dbzObj == 1;
  200.                 } else {
  201.                     return Boolean.parseBoolean(dbzObj.toString());
  202.                 }
  203.             }
  204.  
  205.         };
  206.     }
  207.  
  208.     private static DeserializationRuntimeConverter convertToShort() {
  209.         return new DeserializationRuntimeConverter() {
  210.  
  211.             private static final long serialVersionUID = 1L;
  212.  
  213.             @Override
  214.             public Object convert(Object dbzObj, Schema schema) {
  215.                 return Short.parseShort(dbzObj.toString());
  216.             }
  217.  
  218.         };
  219.     }
  220.  
  221.     private static DeserializationRuntimeConverter convertToInt() {
  222.         return new DeserializationRuntimeConverter() {
  223.             private static final long serialVersionUID = 1L;
  224.  
  225.             @Override
  226.             public Object convert(Object dbzObj, Schema schema) {
  227.                 if (dbzObj instanceof Integer) {
  228.                     return dbzObj;
  229.                 } else if (dbzObj instanceof Long) {
  230.                     return ((Long) dbzObj).intValue();
  231.                 } else {
  232.                     return Integer.parseInt(dbzObj.toString());
  233.                 }
  234.             }
  235.  
  236.         };
  237.     }
  238.  
  239.     private static DeserializationRuntimeConverter convertToLong() {
  240.         return new DeserializationRuntimeConverter() {
  241.             private static final long serialVersionUID = 1L;
  242.  
  243.             @Override
  244.             public Object convert(Object dbzObj, Schema schema) {
  245.                 if (dbzObj instanceof Integer) {
  246.                     return ((Integer) dbzObj).longValue();
  247.                 } else if (dbzObj instanceof Long) {
  248.                     return dbzObj;
  249.                 } else {
  250.                     return Long.parseLong(dbzObj.toString());
  251.                 }
  252.             }
  253.  
  254.         };
  255.     }
  256.  
  257.     private static DeserializationRuntimeConverter createDecimalConverter(DecimalType decimalType) {
  258.         int precision = decimalType.getPrecision();
  259.         int scale = decimalType.getScale();
  260.         if(precision > 38){
  261.             precision = 38;
  262.         }
  263.  
  264.         if(scale > 10){
  265.             scale = 10;
  266.         }
  267.  
  268.         int finalPrecision = precision;
  269.         int finalScale = scale;
  270.         return new DeserializationRuntimeConverter() {
  271.             private static final long serialVersionUID = 1L;
  272.  
  273.             @Override
  274.             public Object convert(Object dbzObj, Schema schema) {
  275.                 BigDecimal bigDecimal;
  276.                 if (dbzObj instanceof byte[]) {
  277.                     // decimal.handling.mode=precise
  278.                     bigDecimal = Decimal.toLogical(schema, (byte[]) dbzObj);
  279.                 } else if (dbzObj instanceof String) {
  280.                     // decimal.handling.mode=string
  281.                     bigDecimal = new BigDecimal((String) dbzObj);
  282.                 } else if (dbzObj instanceof Double) {
  283.                     // decimal.handling.mode=double
  284.                     bigDecimal = BigDecimal.valueOf((Double) dbzObj);
  285.                 } else {
  286.                     if (VariableScaleDecimal.LOGICAL_NAME.equals(schema.name())) {
  287.                         SpecialValueDecimal decimal =
  288.                                 VariableScaleDecimal.toLogical((Struct) dbzObj);
  289.                         bigDecimal = decimal.getDecimalValue().orElse(BigDecimal.ZERO);
  290.                     } else {
  291.                         // fallback to string
  292.                         bigDecimal = new BigDecimal(dbzObj.toString());
  293.                     }
  294.                 }
  295.                 return DecimalData.fromBigDecimal(bigDecimal, finalPrecision, finalScale);
  296.             }
  297.  
  298.         };
  299.     }
  300.  
  301.     private static DeserializationRuntimeConverter convertToDouble() {
  302.         return new DeserializationRuntimeConverter() {
  303.             private static final long serialVersionUID = 1L;
  304.  
  305.             @Override
  306.             public Object convert(Object dbzObj, Schema schema) {
  307.                 if (dbzObj instanceof Float) {
  308.                     return ((Float) dbzObj).doubleValue();
  309.                 } else if (dbzObj instanceof Double) {
  310.                     return dbzObj;
  311.                 } else {
  312.                     return Double.parseDouble(dbzObj.toString());
  313.                 }
  314.             }
  315.  
  316.         };
  317.     }
  318.  
  319.     private static DeserializationRuntimeConverter convertToFloat() {
  320.         return new DeserializationRuntimeConverter() {
  321.             private static final long serialVersionUID = 1L;
  322.  
  323.             @Override
  324.             public Object convert(Object dbzObj, Schema schema) {
  325.                 if (dbzObj instanceof Float) {
  326.                     return dbzObj;
  327.                 } else if (dbzObj instanceof Double) {
  328.                     return ((Double) dbzObj).floatValue();
  329.                 } else {
  330.                     return Float.parseFloat(dbzObj.toString());
  331.                 }
  332.             }
  333.  
  334.         };
  335.     }
  336.  
  337.     private static DeserializationRuntimeConverter convertToDate() {
  338.         return new DeserializationRuntimeConverter() {
  339.             private static final long serialVersionUID = 1L;
  340.  
  341.             @Override
  342.             public Object convert(Object dbzObj, Schema schema) {
  343.                 return TemporalConversions.toLocalDate(dbzObj);
  344.             }
  345.  
  346.         };
  347.     }
  348.  
  349.     private static DeserializationRuntimeConverter convertToTime() {
  350.         return new DeserializationRuntimeConverter() {
  351.             private static final long serialVersionUID = 1L;
  352.  
  353.             @Override
  354.             public Object convert(Object dbzObj, Schema schema) {
  355.                 if (dbzObj instanceof Long) {
  356.                     int seconds = 0;
  357.                     switch (schema.name()) {
  358.                         case MicroTime.SCHEMA_NAME:
  359.                             seconds = (int) ((long) dbzObj / 1000);
  360.                         case NanoTime.SCHEMA_NAME:
  361.                             seconds = (int) ((long) dbzObj / 1000_000);
  362.                     }
  363.                     return LocalTime.ofSecondOfDay(seconds);
  364.                 } else if (dbzObj instanceof Integer) {
  365.                     return LocalTime.ofSecondOfDay((Integer) dbzObj);
  366.                 }
  367.                 // get number of milliseconds of the day
  368.                 return TemporalConversions.toLocalTime(dbzObj);
  369.             }
  370.  
  371.         };
  372.     }
  373.  
  374.     private static DeserializationRuntimeConverter convertToTimestamp(ZoneId serverTimeZone) {
  375.         return new DeserializationRuntimeConverter() {
  376.             private static final long serialVersionUID = 1L;
  377.  
  378.             @Override
  379.             public Object convert(Object dbzObj, Schema schema) {
  380.                 if (dbzObj instanceof Long) {
  381.                     switch (schema.name()) {
  382.                         case Timestamp.SCHEMA_NAME:
  383.                             return TimestampData.fromEpochMillis((Long) dbzObj);
  384.                         case MicroTimestamp.SCHEMA_NAME:
  385.                             long micro = (long) dbzObj;
  386.                             return TimestampData.fromEpochMillis(
  387.                                     micro / 1000, (int) (micro % 1000 * 1000));
  388.                         case NanoTimestamp.SCHEMA_NAME:
  389.                             long nano = (long) dbzObj;
  390.                             return TimestampData.fromEpochMillis(
  391.                                     nano / 1000_000, (int) (nano % 1000_000));
  392.                     }
  393.                 }
  394.                 LocalDateTime localDateTime =
  395.                         TemporalConversions.toLocalDateTime(dbzObj, serverTimeZone);
  396.                 return TimestampData.fromLocalDateTime(localDateTime);
  397.             }
  398.  
  399.         };
  400.     }
  401.  
  402.     private static DeserializationRuntimeConverter convertToLocalTimeZoneTimestamp(ZoneId serverTimeZone) {
  403.         return new DeserializationRuntimeConverter() {
  404.             private static final long serialVersionUID = 1L;
  405.  
  406.             @Override
  407.             public Object convert(Object dbzObj, Schema schema) {
  408.                 if (dbzObj instanceof String) {
  409.                     String str = (String) dbzObj;
  410.                     // TIMESTAMP_LTZ type is encoded in string type
  411.                     Instant instant = Instant.parse(str);
  412.                     return TimestampData.fromLocalDateTime(
  413.                             LocalDateTime.ofInstant(instant, serverTimeZone));
  414.                 }
  415.                 throw new IllegalArgumentException(
  416.                         "Unable to convert to TimestampData from unexpected value '"
  417.                                 + dbzObj
  418.                                 + "' of type "
  419.                                 + dbzObj.getClass().getName());
  420.             }
  421.  
  422.         };
  423.     }
  424.  
  425.     private static DeserializationRuntimeConverter convertToString() {
  426.         return new DeserializationRuntimeConverter() {
  427.             private static final long serialVersionUID = 1L;
  428.  
  429.             @Override
  430.             public Object convert(Object dbzObj, Schema schema) {
  431.                 return StringData.fromString(dbzObj.toString());
  432.             }
  433.  
  434.         };
  435.     }
  436.  
  437.     private static DeserializationRuntimeConverter convertToBinary() {
  438.         return new DeserializationRuntimeConverter() {
  439.             private static final long serialVersionUID = 1L;
  440.  
  441.             @Override
  442.             public Object convert(Object dbzObj, Schema schema) {
  443.                 if (dbzObj instanceof byte[]) {
  444.                     return dbzObj;
  445.                 } else if (dbzObj instanceof ByteBuffer) {
  446.                     ByteBuffer byteBuffer = (ByteBuffer) dbzObj;
  447.                     byte[] bytes = new byte[byteBuffer.remaining()];
  448.                     byteBuffer.get(bytes);
  449.                     return bytes;
  450.                 } else {
  451.                     throw new UnsupportedOperationException(
  452.                             "Unsupported BYTES value type: " + dbzObj.getClass().getSimpleName());
  453.                 }
  454.             }
  455.  
  456.         };
  457.     }
  458.  
  459.     private static DeserializationRuntimeConverter createRowConverter(RowType rowType) {
  460.         final DeserializationRuntimeConverter[] fieldConverters =
  461.                 rowType.getFields().stream()
  462.                         .map(RowType.RowField::getType)
  463.                         .map(MySQLDebeziumDeserializer::createNotNullConverter)
  464.                         .toArray(DeserializationRuntimeConverter[]::new);
  465.         final String[] fieldNames = rowType.getFieldNames().toArray(new String[0]);
  466.  
  467.         return new DeserializationRuntimeConverter() {
  468.             private static final long serialVersionUID = 1L;
  469.  
  470.             @Override
  471.             public Object convert(Object dbzObj, Schema schema) throws Exception {
  472.                 Struct struct = (Struct) dbzObj;
  473.                 int arity = fieldNames.length;
  474.                 Object[] fieldByPosition = new Object[arity];
  475.                 LinkedHashMap<String, Integer> positionByName = new LinkedHashMap<>();
  476.                 for (int i = 0; i < arity; i++) {
  477.                     String fieldName = fieldNames[i];
  478.                     Field field = schema.field(fieldName);
  479.                     positionByName.put(fieldName, i);
  480.                     if (field == null) {
  481.                         fieldByPosition[i] = null;
  482.                     } else {
  483.                         Object fieldValue = struct.getWithoutDefault(fieldName);
  484.                         Schema fieldSchema = schema.field(fieldName).schema();
  485.                         Object convertedField = convertField(fieldConverters[i], fieldValue, fieldSchema);
  486.                         fieldByPosition[i] = convertedField;
  487.                     }
  488.                 }
  489.                 return RowUtils.createRowWithNamedPositions(null, fieldByPosition, positionByName);
  490.             }
  491.  
  492.         };
  493.     }
  494.  
  495.     private static Object convertField(DeserializationRuntimeConverter fieldConverter, Object fieldValue,
  496.                                        Schema fieldSchema) throws Exception {
  497.         if (fieldValue == null) {
  498.             return null;
  499.         } else {
  500.             return fieldConverter.convert(fieldValue, fieldSchema);
  501.         }
  502.     }
  503.  
  504. }
4.自定义Greenplum 写入算子

自定义gp的写入算子，继承Flink RichSinkFunction 针对方法进行实现。

接收Flink收集器传送过来的数据，根据数据的changelog，表名称 自定义生成SQL语句，更新greenplum中的数据

  1. import lombok.extern.slf4j.Slf4j;
  2. import org.apache.commons.collections.CollectionUtils;
  3. import org.apache.commons.lang3.StringUtils;
  4. import org.apache.flink.api.java.tuple.Tuple2;
  5. import org.apache.flink.configuration.Configuration;
  6. import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
  7. import org.apache.flink.table.data.DecimalData;
  8. import org.apache.flink.table.data.StringData;
  9. import org.apache.flink.table.data.TimestampData;
  10. import org.apache.flink.types.Row;
  11. import org.apache.flink.types.RowKind;
  12.  
  13. import java.sql.*;
  14. import java.time.LocalDate;
  15. import java.time.LocalTime;
  16. import java.util.Map;
  17. import java.util.Properties;
  18. import java.util.Set;
  19. import java.util.StringJoiner;
  20. import java.util.stream.Collectors;
  21.  
  22. @Slf4j
  23. public class DwSink extends RichSinkFunction<Tuple2<String, Row>> {
  24.  
  25.     private Connection connection = null;
  26.     private PreparedStatement stmt = null;
  27.     private Map<String, String> tableMapping;
  28.     private String mapper = null;
  29.     private String schema = null;
  30.  
  31.     private CdcPro cdcPro = null;
  32.  
  33.     public DwSink(CdcPro cdcPro) {
  34.         this.cdcPro = cdcPro;
  35.     }
  36.  
  37.     @Override
  38.     public void open(Configuration parameters) throws Exception {
  39.         //读取配置
  40.         Properties properties = PropUtils.getProperties();
  41.         String url = properties.getProperty("dw_jdbc");
  42.         String username = properties.getProperty("dw_user");
  43.         String password = properties.getProperty("dw_password");
  44.         connection = DriverManager.getConnection(url, username, password);
  45.         //初始化的表名映射关系
  46.         String mapping = properties.getProperty("tableMapping");
  47.         //正则匹配
  48.         if (mapping.contains("{tableName}")) {
  49.             mapper = mapping;
  50.         } else {//Map匹配
  51.             this.tableMapping = (Map) JacksonUtil.fromJson(mapping, Map.class);
  52.         }
  53.         schema = properties.getProperty("dw_schema");
  54.         if (StringUtils.isBlank(schema)) {
  55.             schema = "public";
  56.         }
  57.     }
  58.  
  59.     @Override
  60.     public void invoke(Tuple2<String, Row> record, Context context) throws Exception {
  61.         log.debug("invoke:" + record);
  62.         Set<String> fieldNames = record.f1.getFieldNames(true);
  63.         if (CollectionUtils.isEmpty(fieldNames)) {
  64.             log.warn("fieldNames is empty, table:{}", record.f0);
  65.             return;
  66.         }
  67.         if (RowKind.UPDATE_AFTER == record.f1.getKind() || RowKind.INSERT == record.f1.getKind()) {
  68.             doInsert(record, fieldNames);
  69.         } else {
  70.             doDelete(record, fieldNames);
  71.         }
  72.     }
  73.  
  74.     /**
  75.      * 删除的处理
  76.      *
  77.      * @param record     record
  78.      * @param fieldNames fieldNames
  79.      * @throws Exception e
  80.      */
  81.     private void doDelete(Tuple2<String, Row> record, Set<String> fieldNames) throws Exception {
  82.         StringJoiner fieldNamesList = new StringJoiner(" and ");
  83.         for (String fieldName : fieldNames) {
  84.             if (record.f1.getField(fieldName) == null) {
  85.                 fieldNamesList.add(quoteIdentifier(coverColumn(fieldName)) + " is null");
  86.             } else {
  87.                 fieldNamesList.add(quoteIdentifier(coverColumn(fieldName)) + " = ?");
  88.             }
  89.         }
  90.         String sql = "delete from " + this.getMapperTableName(record.f0) + " where " + fieldNamesList;
  91.         log.debug("del sql: {}", sql);
  92.         PreparedStatement stmt = connection.prepareStatement(sql);
  93.         int index = 1;
  94.         for (String fieldName : fieldNames) {
  95.             if (record.f1.getField(fieldName) != null) {
  96.                 handlePreparedStatement(stmt, record.f1.getField(fieldName), index);
  97.                 index++;
  98.             }
  99.         }
  100.         stmt.execute();
  101.     }
  102.  
  103.     /**
  104.      * 插入数据的处理
  105.      *
  106.      * @param record     record
  107.      * @param fieldNames fieldNames
  108.      * @throws Exception e
  109.      */
  110.     private void doInsert(Tuple2<String, Row> record, Set<String> fieldNames) throws Exception {
  111.         StringJoiner fieldNamesList = new StringJoiner(",");
  112.         StringJoiner fieldParmList = new StringJoiner(",");
  113.         for (String fieldName : fieldNames) {
  114.             fieldNamesList.add(quoteIdentifier(coverColumn(fieldName)));
  115.             fieldParmList.add("?");
  116.         }
  117.         String sql = "insert into " + this.getMapperTableName(record.f0) + " ( " + fieldNamesList + " )" +
  118.                 " values (" + fieldParmList + ")";
  119.         log.info("insert sql: {}", sql);
  120.         PreparedStatement stmt = connection.prepareStatement(sql);
  121.         int index = 1;
  122.         for (String fieldName : fieldNames) {
  123.             Object field = record.f1.getField(fieldName);
  124.             handlePreparedStatement(stmt, field, index);
  125.             index++;
  126.         }
  127.         stmt.execute();
  128.         stmt.close();
  129.     }
  130.  
  131.     /**
  132.      * 处理flink数据类型到java-jdbc的数据类型
  133.      *
  134.      * @param stmt  stmt
  135.      * @param field field
  136.      * @param index index
  137.      * @throws Exception e
  138.      */
  139.     @Deprecated
  140.     private void handlePreparedStatement(PreparedStatement stmt, Object field, int index) throws Exception {
  141. //        if (field instanceof BinaryStringData) {
  142. //            //JSON类型的数据序列化过来是BinaryStringData
  143. //            PGobject pGobject = new PGobject();
  144. //            pGobject.setType("json");
  145. //            pGobject.setValue(field.toString());
  146. //            stmt.setObject(index, pGobject);
  147. //        } else
  148.         if (field instanceof StringData) {
  149.             stmt.setString(index, String.valueOf(field));
  150.         } else if (field instanceof DecimalData) {
  151.             stmt.setBigDecimal(index, ((DecimalData) field).toBigDecimal());
  152.         } else if (field instanceof TimestampData) {
  153.             stmt.setTimestamp(index, ((TimestampData) field).toTimestamp());
  154.         } else if (field instanceof Short) {
  155.             stmt.setShort(index, (Short) field);
  156.         } else if (field instanceof Boolean) {
  157.             stmt.setBoolean(index, (Boolean) field);
  158.         } else if (field instanceof Integer) {
  159.             stmt.setInt(index, (Integer) field);
  160.         } else if (field instanceof Long) {
  161.             stmt.setLong(index, (Long) field);
  162.         } else if (field instanceof Double) {
  163.             stmt.setDouble(index, (Double) field);
  164.         } else if (field instanceof Float) {
  165.             stmt.setFloat(index, (Float) field);
  166.         } else if (field instanceof LocalDate) {
  167.             stmt.setDate(index, Date.valueOf((LocalDate) field));
  168.         } else if (field instanceof LocalTime) {
  169.             LocalTime localTime = (LocalTime) field;
  170.             stmt.setTime(index, new Time(localTime.getHour(), localTime.getMinute(), localTime.getSecond()));
  171.         } else if (field instanceof byte[]) {
  172.             stmt.setBytes(index, (byte[]) field);
  173.         } else {
  174.             stmt.setObject(index, field);
  175.         }
  176.     }
  177.  
  178.     private String quoteIdentifier(String identifier) {
  179.         return "\"" + identifier + "\"";
  180.     }
  181.  
  182.     private String coverColumn(String fileName){
  183.         if(StringUtils.isNotEmpty(fileName) && StringUtils.isNotEmpty(cdcPro.getColumnConver())){
  184.  
  185.             if("uppercase".equalsIgnoreCase(cdcPro.getColumnConver())){
  186.                 return fileName.toUpperCase();
  187.             }else if("lowercase".equalsIgnoreCase(cdcPro.getColumnConver())){
  188.                 return fileName.toLowerCase();
  189.             }
  190.         }
  191.         return fileName;
  192.     }
  193.  
  194.     @Override
  195.     public void close() throws Exception {
  196.         if (stmt != null) {
  197.             stmt.close();
  198.         }
  199.         if (connection != null) {
  200.             connection.close();
  201.         }
  202.     }
  203.  
  204.     /**
  205.      * 根据映射规则找到mysql表的对应dw的表
  206.      *
  207.      * @param originalTableName originalTableName
  208.      * @return originalTableName
  209.      */
  210.     private String getMapperTableName(String originalTableName) {
  211.         String dwTableName;
  212.         if (!StringUtils.isBlank(mapper)) {
  213.             dwTableName = mapper.replace("{tableName}", originalTableName);
  214.         } else {
  215.             dwTableName = tableMapping.get(originalTableName);
  216.         }
  217.         if (StringUtils.isBlank(dwTableName)) {
  218.             log.error("mysql表: " + originalTableName + "没有找到对应的dw库所对应的表!");
  219.         }
  220.         return schema + "." + dwTableName;
  221.     }
  222.  
  223. }
5.参数配置
  1. #mysql--source
  2. mysql_jdbc=jdbc:mysql://ip:3306/test?autoReconnect=true
  3. mysql_userName=root
  4. mysql_passWord=123456
  5. #\u7528\u9017\u53F7\u5206\u9694\u8868\u540D\uFF0C\u5982\u679C\u662F\u8BE5\u5E93\u4E0B\u7684\u5168\u90E8\u8868\uFF0C\u5219\u7528.*\u5373\u53EF
  6. mysql_tableList=.*
  7. excluded_tables=t1,t2,table_decimal
  8. #dw--sink
  9. dw_jdbc=jdbc:postgresql://ip:5432/postgres?autoReconnect=true&useUnicode=true&characterEncoding=utf8&serverTimezone=GMT%2B8&useSSL=false
  10. dw_schema=ods_cdctest1
  11. dw_user=gpadmin
  12. dw_password=123456
  13. #\u6620\u5C04\u5173\u7CFB\uFF0C\u53EF\u4EE5\u9009\u62E9map\u6620\u5C04\uFF0C\u4E5F\u53EF\u4EE5\u9009\u62E9\u4E0B\u9762\u6B63\u5219\u6620\u5C04\uFF0C\u4E8C\u9009\u4E00
  14. tableMapping={tableName}
  15. column_conver=lowercase
  16. #\u6216\u8005\u5339\u914D\u6B63\u5219\uFF0C\u6BD4\u5982\uFF1A
  17. #tableMapping=abc_{tableName}
  18.  
  19. copy_dw_schema=public
  20. copy_dw_table=test
其他

如果需要捕获表结构变更，可以实现数据库结构变更事件

  1. public void deserialize(SourceRecord record, Collector out) throws Exception {
  2.         Struct value = (Struct) record.value();
  3.         Schema keySchema = record.keySchema();
  4.         //判断是否是数据库结构变更
  5.         if(keySchema != null && keySchema.name().equals(SourceRecordUtils.SCHEMA_CHANGE_EVENT_KEY_NAME)){
  6.             /**
  7.              * 获取读取的schema 变更信息
  8.              * 类似于
  9.              * {"source":{"file":"mysql-bin.000079","pos":4598526,"server_id":123654},"position":{"transaction_id":null,"ts_sec":1698822218,"file":"mysql-bin.000079","pos":4598686,"server_id":123654},"databaseName":"test1","ddl":"create table table_name1\n(\n    id int null\n)","tableChanges":[]}
  10.              *
  11.              */
  12.             String historyRecord = (String) value.get("historyRecord");
  13.             Map recordMap = (Map) JacksonUtil.fromJson(historyRecord, Map.class);
  14.             String ddl = recordMap.get("ddl").toString();
  15.             String database = recordMap.get("databaseName").toString();
  16.             out.collect(Tuple2.of(database,ddl));
  17.         }

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/weixin_40725706/article/detail/1019879
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号