To resolve dependency conflicts, ensure your Maven configuration is correct. For instance, if the IDE does not reflect database changes, setting paralelism to 1 might help.

Step 5: Transmit Data in JSON Format to Kafka

Debezium-json format captures and converts database change events into JSON messages suitable to message queues like Kafka.

tEnv.executeSql("CREATE TABLE kafka_sink (\n" +
                "  identifier INT,\n" +
                "  fullName STRING,\n" +
                "  years BIGINT,\n" +
                "  gender String,\n" +
                "  PRIMARY KEY (identifier) NOT ENFORCED\n"+
                ")  WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'db_changes',\n" +  
                "  'properties.bootstrap.servers' = 'server1:9092',\n" + 
                "  'properties.group.id' = 'groupA',\n" +
                "  'scan.startup.mode' = 'earliest-offset',\n" +
                "  'format' = 'debezium-json'\n" +  
                ")");
tEnv.executeSql("INSERT INTO kafka_sink SELECT * FROM Employee");

Alternative Approach: Using Flink API

The steps before data fetching are similar to using Flink SQL.

// Configuration properties
Properties configProps = new Properties();
configProps.put("useSSL","false");

MySqlSource<string> dataSource = MySqlSource.<string>builder()
        .hostname("server1")  
        .port(3306)  
        .databaseList("example_db") 
        .tableList("example_db.Employee")
        .username("admin")  
        .password("pwd123") 
        .deserializer(new JsonDebeziumDeserializationSchema()) 
        .jdbcProperties(configProps)  
        .build();

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(3000);

env.fromSource(dataSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
   .setParallelism(1)
   .print();

env.execute("Display MySQL Snapshot + Binlog");
</string></string>

Custom Deserialization Schema

For tailored JSON outputs focusing on specific fields:

public static class RecordDeserializer implements DebeziumDeserializationSchema<string> {
    @Override
    public void deserialize(SourceRecord record, Collector<string> collector) throws Exception {
        String[] topicParts = record.topic().split("\\.");
        String db = topicParts[1], table = topicParts[2];
        
        JSONObject outputData = new JSONObject();
        Struct valueStruct = (Struct) record.value();
        JSONObject beforeJson = new JSONObject(), afterJson = new JSONObject();
        
        Struct beforeStruct = valueStruct.getStruct("before");
        if(beforeStruct != null){
            beforeStruct.schema().fields().forEach(field -> 
                beforeJson.put(field.name(), beforeStruct.get(field)));
        }
        
        Struct afterStruct = valueStruct.getStruct("after");
        afterStruct.schema().fields().forEach(field -> 
            afterJson.put(field.name(), afterStruct.get(field)));
        
        outputData.put("database", db).put("table", table)
                  .put("before", beforeJson).put("after", afterJson);
        
        collector.collect(outputData.toString());
    }

    @Override
    public TypeInformation<string> getProducedType() {
        return BasicTypeInfo.STRING_TYPE_INFO;
    }
}
</string></string></string>