CDC stands for Change Data Capture, which is a broad concept, as long as it can capture the change data, it can be called CDC. Flink CDC is a Log message-based data capture tool, all the inventory and incremental data can be captured. Taking MySQL as an example, it can easily capture Binlog data through Debezium and process the calculations in real time to send them to the data lake. The data lake can then be queried by other engines.

This section will show how to ingest one table or multiple tables into the data lake for both Iceberg format and Mixed-Iceberg format.

Ingest into one table

Iceberg format

The following example will show how MySQL CDC data is written to an Iceberg table.

Requirements

Please add Flink SQL Connector MySQL CDC and Iceberg Jars to the lib directory of the Flink engine package.

  1. CREATE TABLE products (
  2.     id INT,
  3.     name STRING,
  4.     description STRING,
  5.     PRIMARY KEY (id) NOT ENFORCED
  6. ) WITH (
  7.     'connector' = 'mysql-cdc',
  8.     'hostname' = 'localhost',
  9.     'port' = '3306',
  10.     'username' = 'root',
  11.     'password' = '123456',
  12.     'database-name' = 'mydb',
  13.     'table-name' = 'products'
  14. );
  16. CREATE CATALOG iceberg_hadoop_catalog WITH (
  17.     'type'='iceberg',
  18.     'catalog-type'='hadoop',
  19.     'warehouse'='hdfs://nn:8020/warehouse/path',
  20.     'property-version'='1'
  21. );
  23. CREATE TABLE IF NOT EXISTS `iceberg_hadoop_catalog`.`default`.`sample` (
  24.     id INT,
  25.     name STRING,
  26.     description STRING,
  27.     PRIMARY KEY (id) NOT ENFORCED
  28. );
  30. INSERT INTO `iceberg_hadoop_catalog`.`default`.`sample` SELECT * FROM products;

Mixed-Iceberg format

The following example will show how MySQL CDC data is written to a Mixed-Iceberg table.

Requirements

Please add Flink SQL Connector MySQL CDC and Amoro Jars to the lib directory of the Flink engine package.

  1. CREATE TABLE products (
  2.     id INT,
  3.     name STRING,
  4.     description STRING,
  5.     PRIMARY KEY (id) NOT ENFORCED
  6. ) WITH (
  7.     'connector' = 'mysql-cdc',
  8.     'hostname' = 'localhost',
  9.     'port' = '3306',
  10.     'username' = 'root',
  11.     'password' = '123456',
  12.     'database-name' = 'mydb',
  13.     'table-name' = 'products'
  14. );
  16. CREATE CATALOG arctic_catalog WITH (
  17.     'type'='arctic',
  18.     'metastore.url'='thrift://<ip>:<port>/<catalog_name_in_metastore>'
  19. ); 
  21. CREATE TABLE IF NOT EXISTS `arctic_catalog`.`db`.`test_tb`(
  22.     id INT,
  23.     name STRING,
  24.     description STRING,
  25.     PRIMARY KEY (id) NOT ENFORCED
  26. );
  28. INSERT INTO `arctic_catalog`.`db`.`test_tb` SELECT * FROM products;

Ingest Into multiple tables

Iceberg format

The following example will show how to write CDC data from multiple MySQL tables into the corresponding Iceberg table.

Requirements

Please add Flink Connector MySQL CDC and Iceberg dependencies to your Maven project’s pom.xml file.

  1. import com.ververica.cdc.connectors.mysql.source.MySqlSource;
  2. import com.ververica.cdc.connectors.mysql.table.MySqlDeserializationConverterFactory;
  3. import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
  4. import com.ververica.cdc.debezium.table.MetadataConverter;
  5. import com.ververica.cdc.debezium.table.RowDataDebeziumDeserializeSchema;
  6. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  7. import org.apache.flink.api.common.typeinfo.TypeInformation;
  8. import org.apache.flink.api.java.tuple.Tuple2;
  9. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  10. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  11. import org.apache.flink.streaming.api.functions.ProcessFunction;
  12. import org.apache.flink.table.api.*;
  13. import org.apache.flink.table.catalog.*;
  14. import org.apache.flink.table.data.RowData;
  15. import org.apache.flink.table.data.conversion.RowRowConverter;
  16. import org.apache.flink.table.data.utils.JoinedRowData;
  17. import org.apache.flink.table.types.logical.RowType;
  18. import org.apache.flink.util.Collector;
  19. import org.apache.flink.util.OutputTag;
  20. import org.apache.hadoop.conf.Configuration;
  21. import org.apache.iceberg.CatalogProperties;
  22. import org.apache.iceberg.Table;
  23. import org.apache.iceberg.catalog.Catalog;
  24. import org.apache.iceberg.catalog.Namespace;
  25. import org.apache.iceberg.catalog.TableIdentifier;
  26. import org.apache.iceberg.flink.CatalogLoader;
  27. import org.apache.iceberg.flink.TableLoader;
  28. import org.apache.iceberg.flink.sink.FlinkSink;
  29. import org.apache.iceberg.relocated.com.google.common.collect.Maps;
  30. import org.apache.kafka.connect.data.Struct;
  31. import org.apache.kafka.connect.source.SourceRecord;
  32. import java.util.*;
  34. import static com.ververica.cdc.connectors.mysql.table.MySqlReadableMetadata.DATABASE_NAME;
  35. import static com.ververica.cdc.connectors.mysql.table.MySqlReadableMetadata.TABLE_NAME;
  36. import static java.util.stream.Collectors.toMap;
  38. public class MySqlCDC2IcebergExample {
  39.   public static void main(String[] args) throws Exception {
  40.     List<Tuple2<ObjectPath, ResolvedCatalogTable>> pathAndTable = initSourceTables();
  41.     Map<String, RowDataDebeziumDeserializeSchema> debeziumDeserializeSchemas = getDebeziumDeserializeSchemas(pathAndTable);
  42.     MySqlSource<RowData> mySqlSource = MySqlSource.<RowData>builder()
  43.       .hostname("yourHostname")
  44.       .port(yourPort)
  45.       .databaseList("test_db")
  46.       // setting up tables to be captured
  47.       .tableList("test_db.user", "test_db.product")
  48.       .username("yourUsername")
  49.       .password("yourPassword")
  50.       .deserializer(new CompositeDebeziumDeserializationSchema(debeziumDeserializeSchemas))
  51.       .build();
  53.     StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  55.     // enable checkpoint
  56.     env.enableCheckpointing(60000);
  58.     // Split CDC streams by table name
  59.     SingleOutputStreamOperator<Void> process = env
  60.       .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
  61.       .setParallelism(4)
  62.       .process(new SplitCdcStreamFunction(pathAndTable.stream()
  63.           .collect(toMap(e -> e.f0.toString(),
  64.               e -> RowRowConverter.create(e.f1.getResolvedSchema().toPhysicalRowDataType())))))
  65.       .name("split stream");
  67.     // create Iceberg sink and insert into CDC data
  68.     Map<String, String> properties = Maps.newHashMap();
  69.     properties.put(CatalogProperties.WAREHOUSE_LOCATION, "yourWarehouseLocation");
  70.     properties.put(CatalogProperties.URI, "yourThriftUri");
  71.     CatalogLoader catalogLoader = CatalogLoader.hadoop("hadoop_catalog", new Configuration(), properties);
  72.     Catalog icebergHadoopCatalog = catalogLoader.loadCatalog();
  73.     Map<String, TableSchema> sinkTableSchemas = new HashMap<>();
  74.     sinkTableSchemas.put("user", TableSchema.builder().field("id", DataTypes.INT())
  75.       .field("name", DataTypes.STRING()).field("op_time", DataTypes.TIMESTAMP()).build());
  76.     sinkTableSchemas.put("product", TableSchema.builder().field("productId", DataTypes.INT())
  77.       .field("price", DataTypes.DECIMAL(12, 6)).field("saleCount", DataTypes.INT()).build());
  79.     for (Map.Entry<String, TableSchema> entry : sinkTableSchemas.entrySet()) {
  80.       TableIdentifier identifier = TableIdentifier.of(Namespace.of("test_db"), entry.getKey());
  81.       Table table = icebergHadoopCatalog.loadTable(identifier);
  82.       TableLoader tableLoader = TableLoader.fromCatalog(catalogLoader, identifier);
  84.       FlinkSink.forRowData(process.getSideOutput(new OutputTag<RowData>(entry.getKey()){}))
  85.         .tableLoader(tableLoader)
  86.         .table(table)
  87.         .append();
  88.     }
  90.     env.execute("Sync MySQL to the Iceberg table");
  91.   }
  93.   static class CompositeDebeziumDeserializationSchema
  94.     implements DebeziumDeserializationSchema<RowData> {
  96.     private final Map<String, RowDataDebeziumDeserializeSchema> deserializationSchemaMap;
  98.     public CompositeDebeziumDeserializationSchema(
  99.       final Map<String, RowDataDebeziumDeserializeSchema> deserializationSchemaMap) {
  100.       this.deserializationSchemaMap = deserializationSchemaMap;
  101.     }
  103.     @Override
  104.     public void deserialize(final SourceRecord record, final Collector<RowData> out)
  105.       throws Exception {
  106.       final Struct value = (Struct) record.value();
  107.       final Struct source = value.getStruct("source");
  108.       final String db = source.getString("db");
  109.       final String table = source.getString("table");
  110.       if (deserializationSchemaMap == null) {
  111.         throw new IllegalStateException("deserializationSchemaMap can not be null!");
  112.       }
  113.       deserializationSchemaMap.get(db + "." + table).deserialize(record, out);
  114.     }
  116.     @Override
  117.     public TypeInformation<RowData> getProducedType() {
  118.       return TypeInformation.of(RowData.class);
  119.     }
  120.   }
  122.   static class SplitCdcStreamFunction extends ProcessFunction<RowData, Void> {
  123.     private final Map<String, RowRowConverter> converters;
  125.     public SplitCdcStreamFunction(final Map<String, RowRowConverter> converterMap) {
  126.       this.converters = converterMap;
  127.     }
  129.     @Override
  130.     public void processElement(final RowData rowData,
  131.                                final ProcessFunction<RowData, Void>.Context ctx, final Collector<Void> out)
  132.       throws Exception {
  133.       // JoinedRowData like +I{row1=+I(1,2.340000,3), row2=+I(product,test_db)}
  134.       // so rowData.getArity() - 2 is the tableName field index
  135.       final String tableName = rowData.getString(rowData.getArity() - 2).toString();
  136.       ctx.output(new OutputTag<RowData>(tableName) {},
  137.         getField(JoinedRowData.class, (JoinedRowData) rowData, "row1"));
  138.     }
  140.     private static <O, V> V getField(Class<O> clazz, O obj, String fieldName) {
  141.       try {
  142.         java.lang.reflect.Field field = clazz.getDeclaredField(fieldName);
  143.         field.setAccessible(true);
  144.         Object v = field.get(obj);
  145.         return v == null ? null : (V) v;
  146.       } catch (NoSuchFieldException | IllegalAccessException e) {
  147.         throw new RuntimeException(e);
  148.       }
  149.     }
  150.   }
  152.   private static List<Tuple2<ObjectPath, ResolvedCatalogTable>> initSourceTables() {
  153.     List<Tuple2<ObjectPath, ResolvedCatalogTable>> pathAndTable = new ArrayList<>();
  154.     // build table "user"
  155.     Schema userSchema = Schema.newBuilder()
  156.       .column("id", DataTypes.INT().notNull())
  157.       .column("name", DataTypes.STRING())
  158.       .column("op_time", DataTypes.TIMESTAMP())
  159.       .primaryKey("id")
  160.       .build();
  161.     List<Column> userTableCols = Stream.of(
  162.       Column.physical("id", DataTypes.INT().notNull()),
  163.       Column.physical("name", DataTypes.STRING()),
  164.       Column.physical("op_time", DataTypes.TIMESTAMP())).collect(Collectors.toList());
  165.     Schema.UnresolvedPrimaryKey userPrimaryKey = userSchema.getPrimaryKey().orElseThrow(() -> new RuntimeException("table user required pk "));
  166.     ResolvedSchema userResolvedSchema = new ResolvedSchema(userTableCols, Collections.emptyList(), UniqueConstraint.primaryKey(
  167.       userPrimaryKey.getConstraintName(), userPrimaryKey.getColumnNames()));
  168.     ResolvedCatalogTable userTable = new ResolvedCatalogTable(
  169.       CatalogTable.of(userSchema, "", Collections.emptyList(), new HashMap<>()), userResolvedSchema);
  170.     pathAndTable.add(Tuple2.of(new ObjectPath("test_db", "user"), userTable));
  172.     // build table "product"
  173.     Schema productSchema = Schema.newBuilder()
  174.       .column("productId", DataTypes.INT().notNull())
  175.       .column("price", DataTypes.DECIMAL(12, 6))
  176.       .column("saleCount", DataTypes.INT())
  177.       .primaryKey("productId")
  178.       .build();
  179.     List<Column> productTableCols = Stream.of(
  180.       Column.physical("productId", DataTypes.INT().notNull()),
  181.       Column.physical("price", DataTypes.DECIMAL(12, 6)),
  182.       Column.physical("saleCount", DataTypes.INT())).collect(Collectors.toList());
  183.     Schema.UnresolvedPrimaryKey productPrimaryKey = productSchema.getPrimaryKey().orElseThrow(() -> new RuntimeException("table product required pk "));
  184.     ResolvedSchema productResolvedSchema = new ResolvedSchema(productTableCols, Collections.emptyList(), UniqueConstraint.primaryKey(
  185.       productPrimaryKey.getConstraintName(), productPrimaryKey.getColumnNames()));
  186.     ResolvedCatalogTable productTable = new ResolvedCatalogTable(
  187.       CatalogTable.of(productSchema, "", Collections.emptyList(), new HashMap<>()), productResolvedSchema);
  188.     pathAndTable.add(Tuple2.of(new ObjectPath("test_db", "product"), productTable));
  189.     return pathAndTable;
  190.   }
  192.   private static Map<String, RowDataDebeziumDeserializeSchema> getDebeziumDeserializeSchemas(
  193.     final List<Tuple2<ObjectPath, ResolvedCatalogTable>> pathAndTable) {
  194.     return pathAndTable.stream()
  195.       .collect(toMap(e -> e.f0.toString(), e -> RowDataDebeziumDeserializeSchema.newBuilder()
  196.         .setPhysicalRowType(
  197.           (RowType) e.f1.getResolvedSchema().toPhysicalRowDataType().getLogicalType())
  198.         .setUserDefinedConverterFactory(MySqlDeserializationConverterFactory.instance())
  199.         .setMetadataConverters(
  200.           new MetadataConverter[] {TABLE_NAME.getConverter(), DATABASE_NAME.getConverter()})
  201.         .setResultTypeInfo(TypeInformation.of(RowData.class)).build()));
  202.   }
  203. }

Mixed-Iceberg format

The following example will show how to write CDC data from multiple MySQL tables into the corresponding Mixed-Iceberg table.

Requirements

Please add Flink Connector MySQL CDC and Amoro dependencies to your Maven project’s pom.xml file.

  1. import com.netease.arctic.flink.InternalCatalogBuilder;
  2. import com.netease.arctic.flink.table.AmoroTableLoader;
  3. import com.netease.arctic.flink.util.AmoroUtils;
  4. import com.netease.arctic.flink.write.FlinkSink;
  5. import com.netease.arctic.table.TableIdentifier;
  6. import com.ververica.cdc.connectors.mysql.source.MySqlSource;
  7. import com.ververica.cdc.connectors.mysql.table.MySqlDeserializationConverterFactory;
  8. import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
  9. import com.ververica.cdc.debezium.table.MetadataConverter;
  10. import com.ververica.cdc.debezium.table.RowDataDebeziumDeserializeSchema;
  11. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  12. import org.apache.flink.api.common.typeinfo.TypeInformation;
  13. import org.apache.flink.api.java.tuple.Tuple2;
  14. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  15. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  16. import org.apache.flink.streaming.api.functions.ProcessFunction;
  17. import org.apache.flink.table.api.*;
  18. import org.apache.flink.table.catalog.*;
  19. import org.apache.flink.table.data.RowData;
  20. import org.apache.flink.table.data.conversion.RowRowConverter;
  21. import org.apache.flink.table.data.utils.JoinedRowData;
  22. import org.apache.flink.table.types.logical.RowType;
  23. import org.apache.flink.util.Collector;
  24. import org.apache.flink.util.OutputTag;
  25. import org.apache.kafka.connect.data.Struct;
  26. import org.apache.kafka.connect.source.SourceRecord;
  27. import java.util.*;
  28. import java.util.stream.Collectors;
  29. import java.util.stream.Stream;
  31. import static com.ververica.cdc.connectors.mysql.table.MySqlReadableMetadata.DATABASE_NAME;
  32. import static com.ververica.cdc.connectors.mysql.table.MySqlReadableMetadata.TABLE_NAME;
  33. import static java.util.stream.Collectors.toMap;
  35. public class MySqlCDC2AmoroExample {
  36.   public static void main(String[] args) throws Exception {
  37.     List<Tuple2<ObjectPath, ResolvedCatalogTable>> pathAndTable = initSourceTables();
  38.     Map<String, RowDataDebeziumDeserializeSchema> debeziumDeserializeSchemas = getDebeziumDeserializeSchemas(
  39.       pathAndTable);
  40.     MySqlSource<RowData> mySqlSource = MySqlSource.<RowData>builder()
  41.       .hostname("yourHostname")
  42.       .port(yourPort)
  43.       .databaseList("test_db")
  44.       // setting up tables to be captured
  45.       .tableList("test_db.user", "test_db.product")
  46.       .username("yourUsername")
  47.       .password("yourPassword")
  48.       .deserializer(new CompositeDebeziumDeserializationSchema(debeziumDeserializeSchemas))
  49.       .build();
  51.     StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  53.     // enable checkpoint
  54.     env.enableCheckpointing(60000);
  56.     // Split CDC streams by table name
  57.     SingleOutputStreamOperator<Void> process = env
  58.       .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source").setParallelism(4)
  59.       .process(new SplitCdcStreamFunction(pathAndTable.stream()
  60.         .collect(toMap(e -> e.f0.toString(),
  61.           e -> RowRowConverter.create(e.f1.getResolvedSchema().toPhysicalRowDataType())))))
  62.       .name("split stream");
  64.     // create Amoro sink and insert into cdc data
  65.     InternalCatalogBuilder catalogBuilder = InternalCatalogBuilder.builder().metastoreUrl(
  66.       "thrift://<ip>:<port>/<catalog_name_in_metastore>");
  67.     Map<String, TableSchema> sinkTableSchemas = new HashMap<>();
  68.     sinkTableSchemas.put("user", TableSchema.builder().field("id", DataTypes.INT())
  69.       .field("name", DataTypes.STRING()).field("op_time", DataTypes.TIMESTAMP()).build());
  70.     sinkTableSchemas.put("product", TableSchema.builder().field("productId", DataTypes.INT())
  71.       .field("price", DataTypes.DECIMAL(12, 6)).field("saleCount", DataTypes.INT()).build());
  73.     for (Map.Entry<String, TableSchema> entry : sinkTableSchemas.entrySet()) {
  74.       TableIdentifier tableId =
  75.         TableIdentifier.of("yourCatalogName", "yourDatabaseName", entry.getKey());
  76.       AmoroTableLoader tableLoader = AmoroTableLoader.of(tableId, catalogBuilder);
  78.       FlinkSink.forRowData(process.getSideOutput(new OutputTag<RowData>(entry.getKey()){}))
  79.         .flinkSchema(entry.getValue())
  80.         .table(AmoroUtils.loadAmoroTable(tableLoader))
  81.         .tableLoader(tableLoader).build();
  82.     }
  84.     env.execute("Sync MySQL to Mixed-Iceberg table");
  85.   }
  87.   static class CompositeDebeziumDeserializationSchema
  88.     implements DebeziumDeserializationSchema<RowData> {
  90.     private final Map<String, RowDataDebeziumDeserializeSchema> deserializationSchemaMap;
  92.     public CompositeDebeziumDeserializationSchema(
  93.       final Map<String, RowDataDebeziumDeserializeSchema> deserializationSchemaMap) {
  94.       this.deserializationSchemaMap = deserializationSchemaMap;
  95.     }
  97.     @Override
  98.     public void deserialize(final SourceRecord record, final Collector<RowData> out)
  99.       throws Exception {
  100.       final Struct value = (Struct) record.value();
  101.       final Struct source = value.getStruct("source");
  102.       final String db = source.getString("db");
  103.       final String table = source.getString("table");
  104.       if (deserializationSchemaMap == null) {
  105.         throw new IllegalStateException("deserializationSchemaMap can not be null!");
  106.       }
  107.       deserializationSchemaMap.get(db + "." + table).deserialize(record, out);
  108.     }
  110.     @Override
  111.     public TypeInformation<RowData> getProducedType() {
  112.       return TypeInformation.of(RowData.class);
  113.     }
  114.   }
  116.   static class SplitCdcStreamFunction extends ProcessFunction<RowData, Void> {
  117.     private final Map<String, RowRowConverter> converters;
  119.     public SplitCdcStreamFunction(final Map<String, RowRowConverter> converterMap) {
  120.       this.converters = converterMap;
  121.     }
  123.     @Override
  124.     public void processElement(final RowData rowData,
  125.                                final ProcessFunction<RowData, Void>.Context ctx, final Collector<Void> out)
  126.       throws Exception {
  127.       // JoinedRowData like +I{row1=+I(1,2.340000,3), row2=+I(product,test_db)}
  128.       // so rowData.getArity() - 2 is the tableName field index
  129.       final String tableName = rowData.getString(rowData.getArity() - 2).toString();
  130.       ctx.output(new OutputTag<RowData>(tableName) {},
  131.         getField(JoinedRowData.class, (JoinedRowData) rowData, "row1"));
  132.     }
  134.     private static <O, V> V getField(Class<O> clazz, O obj, String fieldName) {
  135.       try {
  136.         java.lang.reflect.Field field = clazz.getDeclaredField(fieldName);
  137.         field.setAccessible(true);
  138.         Object v = field.get(obj);
  139.         return v == null ? null : (V) v;
  140.       } catch (NoSuchFieldException | IllegalAccessException e) {
  141.         throw new RuntimeException(e);
  142.       }
  143.     }
  144.   }
  146.   private static List<Tuple2<ObjectPath, ResolvedCatalogTable>> initSourceTables() {
  147.     List<Tuple2<ObjectPath, ResolvedCatalogTable>> pathAndTable = new ArrayList<>();
  148.     // build table "user"
  149.     Schema userSchema = Schema.newBuilder()
  150.       .column("id", DataTypes.INT().notNull())
  151.       .column("name", DataTypes.STRING())
  152.       .column("op_time", DataTypes.TIMESTAMP())
  153.       .primaryKey("id")
  154.       .build();
  155.     List<Column> userTableCols = Stream.of(
  156.       Column.physical("id", DataTypes.INT().notNull()),
  157.       Column.physical("name", DataTypes.STRING()),
  158.       Column.physical("op_time", DataTypes.TIMESTAMP())).collect(Collectors.toList());
  159.     Schema.UnresolvedPrimaryKey userPrimaryKey = userSchema.getPrimaryKey().orElseThrow(() -> new RuntimeException("table user required pk "));
  160.     ResolvedSchema userResolvedSchema = new ResolvedSchema(userTableCols, Collections.emptyList(), UniqueConstraint.primaryKey(
  161.       userPrimaryKey.getConstraintName(), userPrimaryKey.getColumnNames()));
  162.     ResolvedCatalogTable userTable = new ResolvedCatalogTable(
  163.       CatalogTable.of(userSchema, "", Collections.emptyList(), new HashMap<>()), userResolvedSchema);
  164.     pathAndTable.add(Tuple2.of(new ObjectPath("test_db", "user"), userTable));
  166.     // build table "product"
  167.     Schema productSchema = Schema.newBuilder()
  168.       .column("productId", DataTypes.INT().notNull())
  169.       .column("price", DataTypes.DECIMAL(12, 6))
  170.       .column("saleCount", DataTypes.INT())
  171.       .primaryKey("productId")
  172.       .build();
  173.     List<Column> productTableCols = Stream.of(
  174.       Column.physical("productId", DataTypes.INT().notNull()),
  175.       Column.physical("price", DataTypes.DECIMAL(12, 6)),
  176.       Column.physical("saleCount", DataTypes.INT())).collect(Collectors.toList());
  177.     Schema.UnresolvedPrimaryKey productPrimaryKey = productSchema.getPrimaryKey().orElseThrow(() -> new RuntimeException("table product required pk "));
  178.     ResolvedSchema productResolvedSchema = new ResolvedSchema(productTableCols, Collections.emptyList(), UniqueConstraint.primaryKey(
  179.       productPrimaryKey.getConstraintName(), productPrimaryKey.getColumnNames()));
  180.     ResolvedCatalogTable productTable = new ResolvedCatalogTable(
  181.       CatalogTable.of(productSchema, "", Collections.emptyList(), new HashMap<>()), productResolvedSchema);
  182.     pathAndTable.add(Tuple2.of(new ObjectPath("test_db", "product"), productTable));
  183.     return pathAndTable;
  184.   }
  186.   private static Map<String, RowDataDebeziumDeserializeSchema> getDebeziumDeserializeSchemas(
  187.     final List<Tuple2<ObjectPath, ResolvedCatalogTable>> pathAndTable) {
  188.     return pathAndTable.stream()
  189.       .collect(toMap(e -> e.f0.toString(), e -> RowDataDebeziumDeserializeSchema.newBuilder()
  190.         .setPhysicalRowType(
  191.           (RowType) e.f1.getResolvedSchema().toPhysicalRowDataType().getLogicalType())
  192.         .setUserDefinedConverterFactory(MySqlDeserializationConverterFactory.instance())
  193.         .setMetadataConverters(
  194.           new MetadataConverter[] {TABLE_NAME.getConverter(), DATABASE_NAME.getConverter()})
  195.         .setResultTypeInfo(TypeInformation.of(RowData.class)).build()));
  196.   }
  197. }