CREATE TABLE source_tenant_employees  (
                                     id STRING,
                                     created_time BIGINT,
                                     email STRING,
                                     hired_date VARCHAR,
                                     job_number STRING,
                                     last_updated_time VARCHAR,
                                     leave_note STRING,
                                     leave_operator_created_time VARCHAR,
                                     leave_operator_id STRING,
                                     leave_operator_name STRING,
                                     leave_time VARCHAR,
                                     master_slave_type SMALLINT,
                                     occupation_type SMALLINT,
                                     `position` STRING,
                                     positive_date VARCHAR,
                                     salary STRING,
                                     salary_bank_number STRING,
                                     status SMALLINT,
                                     work_place STRING,
                                     organization_id STRING,
                                     superior_id STRING,
                                     tenant_user_id STRING,
                                     unique_offset STRING,
                                     department_id STRING,
                                     insure SMALLINT,
                                     is_domain_admin SMALLINT,
                                     identification STRING,
                                     interview STRING,
                                     person_status SMALLINT,
                                     plan_positive_date VARCHAR,
                                     probation STRING,
                                     hired_operator_created_time VARCHAR,
                                     hired_operator_id STRING,
                                     hired_operator_name STRING,
                                     last_sync_time VARCHAR,
                                     PRIMARY KEY (id) NOT ENFORCED
)
with(
    'connector'='jdbc',
    'url'='jdbc:mysql://49.4.21.141:45611/obpm2',
    'driver'='com.mysql.cj.jdbc.Driver',
    'username'='bcx',
    'password'='Wstestv5qy#2022',
    'table-name'='tenant_employees'
    );
CREATE TABLE sink_tenant_employees  (
                                     id STRING,
                                     created_time VARCHAR,
                                     email STRING,
                                     hired_date VARCHAR,
                                     job_number STRING,
                                     last_updated_time VARCHAR,
                                     leave_note STRING,
                                     leave_operator_created_time VARCHAR,
                                     leave_operator_id STRING,
                                     leave_operator_name STRING,
                                     leave_time VARCHAR,
                                     master_slave_type SMALLINT,
                                     occupation_type SMALLINT,
                                     `position` STRING,
                                     positive_date VARCHAR,
                                     salary STRING,
                                     salary_bank_number STRING,
                                     status SMALLINT,
                                     work_place STRING,
                                     organization_id STRING,
                                     superior_id STRING,
                                     tenant_user_id STRING,
                                     unique_offset STRING,
                                     department_id STRING,
                                     insure SMALLINT,
                                     is_domain_admin SMALLINT,
                                     identification STRING,
                                     interview STRING,
                                     person_status SMALLINT,
                                     plan_positive_date VARCHAR,
                                     probation STRING,
                                     hired_operator_created_time VARCHAR,
                                     hired_operator_id STRING,
                                     hired_operator_name STRING,
                                     last_sync_time VARCHAR,
                                     PRIMARY KEY (id) NOT ENFORCED
)
    with(
        'connector'='jdbc',
        'url'='jdbc:mysql://114.115.253.155:16823/obpm2',
        'driver'='com.mysql.cj.jdbc.Driver',
        'username'='root',
        'password'='Dxdf#0519.com',
        'table-name'='tenant_employees'
        );
insert into sink_tenant_employees select * from source_tenant_employees;