bootstrap.servers=localhost:9092
# unique name for the cluster, used in forming the Connect cluster group. Note that this must not conflict with consumer group IDs
group.id=connect-cluster
# The converters specify the format of data in Kafka and how to translate it into Connect data. Every Connect user will
# need to configure these based on the format they want their data in when loaded from or stored into Kafka
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
# Converter-specific settings can be passed in by prefixing the Converter's setting with the converter we want to apply
key.converter.schemas.enable=true
value.converter.schemas.enable=true
# Topic to use for storing offsets. This topic should have many partitions and be replicated and compacted.
# Kafka Connect will attempt to create the topic automatically when needed, but you can always manually create
# the topic before starting Kafka Connect if a specific topic configuration is needed.
# Most users will want to use the built-in default replication factor of 3 or in some cases even specify a larger value.
# Since this means there must be at least as many brokers as the maximum replication factor used, we'd like to be able
# to run this example on a single-broker cluster and so here we instead set the replication factor to 1.
offset.storage.topic=connect-offsets
# Topic to use for storing connector and task configurations; note that this should be a single partition, highly replicated,
# and compacted topic. Kafka Connect will attempt to create the topic automatically when needed, but you can always manually create
# the topic before starting Kafka Connect if a specific topic configuration is needed.
# Most users will want to use the built-in default replication factor of 3 or in some cases even specify a larger value.
# Since this means there must be at least as many brokers as the maximum replication factor used, we'd like to be able
# to run this example on a single-broker cluster and so here we instead set the replication factor to 1.
config.storage.topic=connect-configs
# Topic to use for storing statuses. This topic can have multiple partitions and should be replicated and compacted.
# Kafka Connect will attempt to create the topic automatically when needed, but you can always manually create
# the topic before starting Kafka Connect if a specific topic configuration is needed.
# Most users will want to use the built-in default replication factor of 3 or in some cases even specify a larger value.
# Since this means there must be at least as many brokers as the maximum replication factor used, we'd like to be able
# to run this example on a single-broker cluster and so here we instead set the replication factor to 1.
status.storage.topic=connect-status

SQL
MySQL [(none)]> show binary logs;
| mysql-bin-changelog.094531 |    176317 |
| mysql-bin-changelog.094532 |    191443 |
| mysql-bin-changelog.094533 |   1102466 |
| mysql-bin-changelog.094534 |    273347 |
| mysql-bin-changelog.094535 |    141555 |
| mysql-bin-changelog.094536 |      4808 |
| mysql-bin-changelog.094537 |    146217 |
| mysql-bin-changelog.094538 |     29607 |
| mysql-bin-changelog.094539 |    141260 |
+----------------------------+-----------+
MySQL [(none)]> show binlog events in 'mysql-bin-changelog.094539';
MySQL [(none)]> show binlog events in 'mysql-bin-changelog.094539' limit 10;
+----------------------------+-----+----------------+------------+-------------+---------------------------------------------------------------------------+
| Log_name                   | Pos | Event_type     | Server_id  | End_log_pos | Info                                                                      |
+----------------------------+-----+----------------+------------+-------------+---------------------------------------------------------------------------+
| mysql-bin-changelog.094539 |   4 | Format_desc    | 1399745413 |         123 | Server ver: 5.7.31-log, Binlog ver: 4                                     |
| mysql-bin-changelog.094539 | 123 | Previous_gtids | 1399745413 |         194 | 90710e1c-f699-11ea-85c0-0ec6a6bed381:1-108842347                          |

$ ./bin/kafka-console-producer.sh -bootstrap-server localhost:9092 --topic  connect-offsets --property "parse.key=true" --property "key.separator=>"
$>["test_servername",{"server":"test_servername"}]>{"ts_sec":1647845014,"file":"mysql-bin-changelog.007051","pos":74121553,"row":1,"server_id":1404217221,"event":2}

{
    "name":"test_servername",
    "config":{
        "connector.class":"io.debezium.connector.mysql.MySqlConnector",
        "snapshot.locking.mode":"none",
        "database.user":"db_user",
        "transforms.Reroute.type":"io.debezium.transforms.ByLogicalTableRouter",
        "database.server.id":"1820615119",
        "database.history.kafka.bootstrap.servers":"localhost:9092",
        "database.history.kafka.topic":"history-topic",
        "inconsistent.schema.handling.mode":"skip",
        "transforms":"Reroute", // 配置binlog数据转发到一个topic，默认一个表一个topic
        "database.server.name":"test_servername",
        "transforms.Reroute.topic.regex":"test_servername(.*)",
        "database.port":"3306",
        "include.schema.changes":"true",
        "transforms.Reroute.topic.replacement":"binlog_data_topic",
        "table.exclude.list":"table_test",
        "database.hostname":"host",
        "database.password":"******",
        "name":"test_servername",
        "database.whitelist":"test_db",
        "database.include.list":"test_db",
        "snapshot.mode":"schema_only_recovery"  // 使用recovery模式从指定binlog文件的offset同步
    }
}

# 写入数据到ODS Raw表
insert_hudi_raw_query = '''
INSERT INTO 
{0}_ods_raw.{1}
SELECT 
{2}
FROM 
{0}_debezium_kafka.kafka_rds_{1}_log
WHERE 
REGEXP(GET_JSON_OBJECT(payload, '$.source.table'), '^{3}$') 
AND GET_JSON_OBJECT(payload, '$.source.db') = '{4}' 
AND IF(GET_JSON_OBJECT(payload, \'$.op\') = \'d\', GET_JSON_OBJECT(payload, \'$.before.{5}\'), GET_JSON_OBJECT(payload, \'$.after.{5}\')) IS NOT NULL
AND GET_JSON_OBJECT(payload, '$.op') IN ('d', 'c', 'u')
'''.format(
    database_name, 
    table_name, 
    hudi_schema, 
    mysql_table_name, 
    mysql_database_name,
    primary_key
)

sh  run_sync_tool.sh  --jdbc-url jdbc:hive2:\/\xxxx:10000 --user hive --pass hive --partitioned-by partition --partition-value-extractor  org.apache.hudi.hive.SlashEncodedHourPartitionValueExtractor --base-path s3://xxx/raw/order_business_db/ord_basics  --auto-create-database  --database order_business_db_ods_raw_hive_sync  --table ord_basics

sh  run_sync_tool.sh  --jdbc-url jdbc:hive2:\/\/xxxx10000 --user hive --pass hive --partitioned-by partition --partition-value-extractor  org.apache.hudi.hive.KlookEncodedDayPartitionValueExtractor --base-path s3://xxxx/raw/order_business_db/ord_basics  --auto-create-database  --database order_business_db_ods_raw_hive_sync  --table ord_basics