env {
  execution.parallelism = 1
  job.mode = "BATCH"
  checkpoint.interval = 5000
  #execution.checkpoint.data-uri = "hdfs://localhost:9000/checkpoint"
}

source {
  jdbc{
    result_table_name = "fake1"
    url = "jdbc:mysql://localhost:3306/digital_operation"
    driver = "com.mysql.cj.jdbc.Driver"
    connection_check_timeout_sec = 100
    user = "root"
    password = "123456"
    query = "select * from test_copy"
  }
}

transform {
    ReplaceRegular {
        source_table_name = "fake"
        result_table_name = "fake1"
        column = "address"
        defaultValue = ""
        conditions = [
            {
               # 值替换的匹配条件
               "compareValue":"郑",
               # 被替换的值
               "replacement":"",
               # 是否包含
               "isContains":true,
               # 是否覆盖
               "isCover":false,
            },{
                # 值替换的匹配条件
                "compareValue":"null",
                # 被替换的值
                "replacement":"郑州",
                # 是否包含
                "isContains":false,
                # 是否替换第一个值
                "isCover":false,
            }
        ]
    }
}

sink {
  jdbc {
    source_table_name = "fake"
    url = "jdbc:mysql://localhost:3306/testdb"
    driver = "com.mysql.cj.jdbc.Driver"
    user = "root"
    password = "123456"
    query = "insert into test_copy(id, name, address, sex, age, create_time, tell, grade, status) values(?, ?, ?, ?, ?, ?, ?, ?, ?)"
    batch_size = 1000
  }
}
