mysql增量同步至hive

import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.format.Json;

public class MySQLToHiveSync {
    public static void main(String[] args) {
        String connectorUrl = "jdbc:mysql://localhost:3306/mydatabase";
        String username = "user";
        String password = "password";
        String databaseName = "mydatabase";

        DebeziumEngine<RecordChangeEvent<MyTable>> engine = DebeziumEngine.create(Json.class)
            .using(getProperties(connectorUrl, username, password, databaseName))
            .notifying(recordChangeEvent -> {
                // 处理同步数据，将数据写入Hive
                System.out.println(recordChangeEvent.record());
            })
            .build();

        // 启动同步引擎
        engine.run();
    }

    private static Properties getProperties(String connectorUrl, String username, String password, String databaseName) {
        Properties props = new Properties();
        props.setProperty("connector.class", "io.debezium.connector.mysql.MySqlConnector");
        props.setProperty("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore");
        props.setProperty("offset.storage.file.filename", "/path/to/storage/offset.dat");
        props.setProperty("offset.storage.topic", "dbhistory.mydatabase");
        props.setProperty("database.hostname", "localhost");
        props.setProperty("database.port", "3306");
        props.setProperty("database.user", username);
        props.setProperty("database.password", password);
        props.setProperty("database.server.id", "184054");
        props.setProperty("database.server.name", "dbserver1");
        props.setProperty("database.include.list", databaseName);
        props.setProperty("database.history.kafka.bootstrap.servers", "kafka:9092");
        props.setProperty("database.history.kafka.topic", "schema-changes.mydatabase");
        return props;
    }
}