/data/tdsql-project/apache-seatunnel-2.3.8/
├── bin/                      # Directory containing execution scripts
├── config/                   # Configuration file directory, including log components, jvm configuration, etc.
├── connectors/               # Directory for storing dependent jar packages
├── lib/                      # Directory for storing dependent jar packages
├── licenses/                 # Directory for storing dependent jar packages 
├── plugins/                  # Directory for storing dependent jar packages
└── starter/                  # Directory for storing dependent jar packages        
    ├── logging/

docker pull apache/kafka:4.1.0
docker run -p 9092:9092 apache/kafka:4.1.0

env {
  parallelism = 1 # Specifies the number of concurrent threads for Source and Sink operations. The actual concurrency may be determined by the Source and Sink configurations.
  job.mode = "BATCH"
  read_limit.bytes_per_second = 7000000 # Rate limiting based on bytes read. If not set, no rate limiting is applied.
  read_limit.rows_per_second = 200000 # Rate limiting based on rows read. If not set, no rate limiting is applied.
} 

CREATE TABLE IF NOT EXISTS `{TableName}`
(
    `K` varbinary(1024) not null,
    `Q` varbinary(256) not null,
    `T` bigint not null,
    `V` mediumblob,
    PRIMARY KEY (`K`, `Q`, `T`))
partition by key(`k`) partitions 6;

sink {
    jdbc {
        # The url must enable bulkload and include rewriteBatchedStatements=true; otherwise, only single insert statements will be executed, resulting in extremely poor import efficiency.
        url = "jdbc:mysql://{ip}:{port}/?rewriteBatchedStatements=true&sessionVariables=tdsql_bulk_load='ON'"        
        driver = "com.mysql.cj.jdbc.Driver"
        
        user = "test"
        password = "test123"
        schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"
        enable_upsert = false           # Disable upsert
        max_retries = 3                 # Number of retries
        generate_sink_sql = "true"      # Generate import SQL based on the database
        
        database = "hbase"              # Import tables to the hbase database
        batch_size = 500000             # The batch transaction size for the tool. To improve efficiency, it accumulates the specified quantity before flushing.
                                        # The larger the better, without exceeding the memory limit of the machine where the tool resides.
        batch_interval_ms = 500
        transaction_timeout = 60000     # Set transaction timeout to 60 seconds (to avoid timeout for large batches)
    }
}

sink {
    LocalFile {
    # Do not modify:
    file_format_type = "text"         # Export file format, currently only text is supported
    field_delimiter = ","             # Field delimiter, only comma is supported
    row_delimiter = "\\n"              # Row delimiter, only newline is supported
    is_enable_transaction = false     # Transaction not supported
    
    # Required:
    path = "/root/dumper_test/apache-seatunnel-2.3.4"     # Export path
    tmp_path = "/root/dumper_test/apache-seatunnel-2.3.4" # During export, data is first exported to tmp_path, then moved to path. Therefore, tmp_path must be set; otherwise, it will occupy the root directory space.
    result_table_name = "test_mytable"                    # Export table name
    result_database_name = "test_mydatabase"              # Export database name, used to generate the file name
    result_column_names = ["id", "name"]                  # Exported column names, must be the same as the number of columns in the source s chema
    
    # Optional:
    parent_directory = "test_directory"         # The parent directory of the actual exported files. If specified, the actual file path will be :path/parent_directory/file; if not specified, the actual path will be :path/file
    file_start_index = 1000                     # Starting sequence for filenames; the sequence in filenames will increment from this number, defaults to 0
  }
}

# The initial value in the configuration file is set too low, which may cause tasks to fail to work under high concurrency due to insufficient memory resource allocation. Adjust the value appropriately based on the machine resources.
-Xms32g  # Specifies the initial heap memory size for the JVM as 16GB
-Xmx128G

set persist tdsql_bulk_load_allow_auto_organize_txn = on;   // Enable bulkload self-organized transaction import; this needs to be disabled for incremental migration
set persist tdsql_bulk_load_allow_unsorted = on;            // Allow bulkload out-of-order transactions
set persist tdsql_bulk_load_commit_threshold = 4294967296;  // commit threshold adjustment
set persist tdsql_bulk_load_rpc_timeout = 7200000;          // rpc timeout

nohup ./bin/seatunnel.sh --config ./config/hbase_config -m local > seatunnel_hbase_local.log 2>&1 &

tail -f seatunnel_hbase_local.log

HBaseDataValidation {
    parallelism = 10
    specified_namespace = "default"
  	table_match_rules = []
    zookeeper_quorum = "127.0.0.1:2181,127.0.0.2:2181,127.0.0.3:2181"
    // Replace with your own username and password, ip port
    mysql_jdbc_url = "jdbc:mysql://{ip}:{port}/hbase?user=test&password=test1234" 
}

./bin/validateKV.sh ./config/validation_config

## dir tree
./apache-seatunnel-2.3.8/bin/
├── install-plugin.cmd
├── install-plugin.sh
├── seatunnel-cluster.cmd
├── seatunnel-cluster.sh
├── seatunnel.cmd
├── seatunnel-connector.cmd
├── seatunnel-connector.sh
├── seatunnel.sh
└── validateKV.sh

# Verify that the replication_scope attribute of the column family in the table to be synchronized is 1.
hbase shell
describe 'your_table_name'

docker pull apache/kafka:4.1.0
docker run -p 9092:9092 apache/kafka:4.1.0

env {
  parallelism = 4
  job.mode = "STREAMING" // Incremental mode requires setting to STREAMING mode
  job.retry.times = 0
}

source {
  Hbase {
    HbaseCDCConfig {
       auto_enable_replication = true
       kafka_bootstrap_servers = "localhost:9092" // If started using docker, keep the default 
       peer_name = "incr_001" // hbase peer name
       // The following ports are used by the self-started hbase of the incremental migration tool. If port conflicts occur, you can configure the ports manually; otherwise, the default ports 2181, 16000, 16010, 16020, 16030 are used.
       zk_host = "ip:12181"
       hbase.master.port = 26000
       hbase.master.info.port = 26010
       hbase.regionserver.port = 26020
       hbase.regionserver.info.port = 26030
    }
    SourceModeKV = {
     specified_namespace = "***" // Specify the namespace to be exported
     table_match_rules = ["***"] // Specify the matching rules for tables to be exported

     dump_create_schema = true
     dump_source_split_metrics = false
    }

    zookeeper_quorum = "127.0.0.1:2181"
    
    # Optional configuration.
    is_kerberos_connection=true # Whether kerberos authentication is required; defaults to false. The following four items only need to be provided when this is set to true.
    kerberos_ms_name="xxx"      # hbase.master.kerberos.principal
    kerberos_rs_name="xxx"      # hbase.regionserver.kerberos.principal
    kerberos_user_name="xxx"
    kerberos_keytab="xxx"
  }
}

sink {
  jdbc {
    // Enable bulk_load
    url = "jdbc:mysql://127.0.0.1:6050/?rewriteBatchedStatements=true&sessionVariables=tdsql_bulk_load='ON'"
    driver = "com.mysql.cj.jdbc.Driver"
    
    properties {
      autoReconnect = "true"
      failOverReadOnly = "false"
      maxReconnects = "10"
      tcpKeepAlive = "true"
      rewriteBatchedStatements = "true"
    }

    user = "test"
    password = "test123"
    schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"
    enable_upsert = true
    max_retries = 1
    generate_sink_sql = true

    database = "hbase" // Import to the hbase database
    batch_size = 500000
  }
}

# The initial value in the configuration file is set too low, which may cause tasks to fail to work under high concurrency due to insufficient memory resource allocation. Adjust the value appropriately based on the machine resources.
-Xms32g  # Specifies the initial heap memory size for the JVM as 16GB
-Xmx128G

set persist tdsql_bulk_load_allow_unsorted = on;            // Allow bulkload out-of-order transactions
set persist tdsql_bulk_load_commit_threshold = 4294967296;  // commit threshold adjustment
set persist tdsql_bulk_load_rpc_timeout = 7200000;          // rpc timeout

./bin/seatunnel.sh --config ./config/incremental_config -m local

HBaseDataValidation {
    parallelism = 10
    specified_namespace = "default" // namespace in hbase
  	table_match_rules = ["(?!.*snapshot_for_seatunnel_incremental_source).*"] // Tables to be validated, filtering out snapshot tables created by the validation tool
    zookeeper_quorum = "127.0.0.1:2181,127.0.0.2:2181,127.0.0.3:2181"
    // Replace with your own username and password, ip port
    mysql_jdbc_url = "jdbc:mysql://{ip}:{port}/hbase?user=test&password=test1234" 
}

./bin/validateKV.sh ./config/validation_config