/data/tdsql-project/apache-seatunnel-2.3.8/
├── bin/                      # 存放运行脚本的目录
├── config/                   # 配置文件目录，包括日志组件，jvm 配置等
├── connectors/               # 存放依赖的jar包
├── lib/                      # 存放依赖的jar包
├── licenses/                 # 存放依赖的jar包 
├── plugins/                  # 存放依赖的jar包
└── starter/                  # 存放依赖的jar包        
    ├── logging/

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

env {
  parallelism = 1 # 指定 Source Sink 时的并发线程数，实际的并发度可能由 Source Sink 配置决定
  job.mode = "BATCH"
  read_limit.bytes_per_second = 7000000 # 按照读取 bytes 限速，不填则不限速
  read_limit.rows_per_second = 200000 # 按照读取行数限速，不填则不限速
} 

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 {
        # url 里需要开启 bulkload，必须打开 rewriteBatchedStatements=true，否则都是单条 insert，导入效率极差
        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           # 不开启 upsert
        max_retries = 3                 # 重试次数
        generate_sink_sql = "true"      # 根据 DB 生成导入SQL
        
        database = "hbase"              # 需要把表导入到hbase db
        batch_size = 500000             # 工具攒批事务的大小，为了效率会攒够指定数量在往下刷
                                        # 在不超过工具所在机器内存限制的情况下，越大越好
        batch_interval_ms = 500
        transaction_timeout = 60000     # 事务超时设为60秒（避免大批次超时）
    }
}

sink {
    LocalFile {
    # 不可修改：
    file_format_type = "text"         # 导出文件格式，目前仅支持 text
    field_delimiter = ","             # 数据分隔符，仅支持逗号
    row_delimiter = "\\n"              # 每行分隔符，仅支持换行符
    is_enable_transaction = false     # 不支持事务
    
    # 必填：
    path = "/root/dumper_test/apache-seatunnel-2.3.4"     # 导出路径
    tmp_path = "/root/dumper_test/apache-seatunnel-2.3.4" # 导出时会先导出到 tmp_path，再 mv 到 path，因此必须设置 tmp_path，否则会占用根目录空间
    result_table_name = "test_mytable"                    # 导出表名
    result_database_name = "test_mydatabase"              # 导出数据库名，用于生成文件名
    result_column_names = ["id", "name"]                  # 导出列的列名，必须和 source 中s chema 里的 columns 数量相同
    
    # 选填：
    parent_directory = "test_directory"         # 实际导出文件的父目录，即如果填写，则文件真实目录为:path/parent_directory/file，如果不填，则真实目录为:path/file
    file_start_index = 1000                     # 文件名的起始序列，文件名中的序列会从此数字开始递增，默认为0
  }
}

# 配置文件里的初始值给的很小，在高并发下可能会因为内存资源调用不足导致无法进行工作，可根据机器资源大小适当给就行
-Xms32g  #指定 JVM 初始堆内存大小为 16GB
-Xmx128G

set persist tdsql_bulk_load_allow_auto_organize_txn = on;   // 允许 bulkload自组织事务导入，增量迁移下需要关闭
set persist tdsql_bulk_load_allow_unsorted = on;            // 允许 bulkload 事务乱序
set persist tdsql_bulk_load_commit_threshold = 4294967296;  // commit 阈值调整
set persist tdsql_bulk_load_rpc_timeout = 7200000;          // rpc 超时时间

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"
    // 自行替换用户名和密码，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

# 确认待同步表的列簇 replication_scope 属性为 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" // 增量模式需要配置为STREAMING模式
  job.retry.times = 0
}

source {
  Hbase {
    HbaseCDCConfig {
       auto_enable_replication = true
       kafka_bootstrap_servers = "localhost:9092" // 如果是使用docker启动，保持默认即可 
       peer_name = "incr_001" // hbase peer name
       // 以下为增量迁移工具自启动的hbase，如端口冲突可自行配置端口，否则默认使用2181 16000 16010 16020 16030端口
       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 = "***" // 指定需要导出的 namespace
     table_match_rules = ["***"] // 指定需要导出的表的匹配规则

     dump_create_schema = true
     dump_source_split_metrics = false
    }

    zookeeper_quorum = "127.0.0.1:2181"
    
    # 可选配置
    is_kerberos_connection=true # 是否需要kerberos验证，默认false。仅当该项为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 {
    // 开启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" // 导入到hbase库
    batch_size = 500000
  }
}

# 配置文件里的初始值给的很小，在高并发下可能会因为内存资源调用不足导致无法进行工作，可根据机器资源大小适当给就行
-Xms32g  #指定 JVM 初始堆内存大小为 16GB
-Xmx128G

set persist tdsql_bulk_load_allow_unsorted = on;            // 允许 bulkload 事务乱序
set persist tdsql_bulk_load_commit_threshold = 4294967296;  // commit 阈值调整
set persist tdsql_bulk_load_rpc_timeout = 7200000;          // rpc 超时时间

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

HBaseDataValidation {
    parallelism = 10
    specified_namespace = "default" // hbase中的namespace
  	table_match_rules = ["(?!.*snapshot_for_seatunnel_incremental_source).*"] // 需要校验的表,过滤掉校验工具创建的快照表
    zookeeper_quorum = "127.0.0.1:2181,127.0.0.2:2181,127.0.0.3:2181"
    // 自行替换用户名和密码，ip port
    mysql_jdbc_url = "jdbc:mysql://{ip}:{port}/hbase?user=test&password=test1234" 
}

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