SET 'execution.checkpointing.interval' = '30s';
/**
  来自kafka的职员信息
 */
CREATE TABLE source_kafka_tenant_employees(
                                          id STRING,created_time BIGINT,tenant_user_id STRING,status SMALLINT,occupation_type SMALLINT,insure_status SMALLINT,
                                          organization_id STRING,department_id STRING,hired_date BIGINT,leave_time BIGINT,
                                          `event_time` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp',
                                          PRIMARY KEY (id) NOT ENFORCED
)WITH (
        'connector' = 'kafka',
        'topic' = 'obpm2.binlog-cdc.topic.v2.tenant_employees',
        'properties.bootstrap.servers' = '117.78.39.204:9092',
        'properties.group.id' = 'kafka-starrock-client',
        'scan.startup.mode' = 'earliest-offset',
        'format' = 'debezium-json'
);

/**
  来自kafka的企业信息
 */
CREATE TABLE source_kafka_tenant_organizations(
    id STRING,name STRING,supervise_depart_id STRING,supervise_depart_name STRING,industry_code STRING,
    institutional_code STRING,supervise_region_code STRING,created_time BIGINT,
    `event_time` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp',
    PRIMARY KEY (id) NOT ENFORCED
)
    WITH (
        'connector' = 'kafka',
        'topic' = 'obpm2.binlog-cdc.topic.v2.tenant_organizations',
        'properties.bootstrap.servers' = '117.78.39.204:9092',
        'properties.group.id' = 'kafka-starrock-client',
        'scan.startup.mode' = 'earliest-offset',
        'format' = 'debezium-json'
        );


CREATE TABLE IF NOT EXISTS sink_dwd_employees (
    id STRING,
    employee_id STRING,
    tenant_user_id STRING,
    occupation_type INT,
    hire_date TIMESTAMP(3),
    leave_date TIMESTAMP(3),
    duty_status INT,
    insure_status INT,
    company_id STRING,
    company_name STRING,
    company_office_type STRING,
    company_supervise_office STRING,
    company_supervise_office_code STRING,
    company_industry_code STRING,
    company_institutional_code STRING,
    company_supervise_region_code STRING,
    department_json STRING,
    att_site_json STRING,
    security_certificate_no STRING,
    level_certificate_no STRING,
    PRIMARY KEY (id) NOT ENFORCED
    ) with (
          'connector' = 'starrocks',
          'jdbc-url' = 'jdbc:mysql://127.0.0.1:9030',
          'load-url' = '127.0.0.1:8030',
          'database-name' = 'olap_dwd',
          'table-name' = 'dwd_employees',
          'username' = 'root',
          'password' = '',
          'sink.buffer-flush.interval-ms' = '5000',
          'sink.properties.column_separator' = '\x01',
          'sink.properties.row_delimiter' = '\x02',
          'sink.properties.partial_update'='true'
 );

/**
  同步职员信息
 */
INSERT INTO sink_dwd_employees
(
    id,employee_id,tenant_user_id,occupation_type,
    hire_date,leave_date,duty_status,insure_status,company_id,company_name,
    company_office_type,company_supervise_office,company_supervise_office_code,
    company_industry_code,company_institutional_code,company_supervise_region_code,
    department_json,att_site_json,security_certificate_no,level_certificate_no
)
SELECT
    c.id,c.id,c.tenant_user_id,c.occupation_type,TO_TIMESTAMP(FROM_UNIXTIME(c.hired_date/1000, 'yyyy-MM-dd HH:mm:ss')),
    TO_TIMESTAMP(FROM_UNIXTIME(c.hired_date/1000, 'yyyy-MM-dd HH:mm:ss')),c.status,c.insure_status,c.organization_id,
    'company_name','company_office_type','company_supervise_office','company_supervise_office_code','company_industry_code',
    'company_institutional_code','company_supervise_region_code','department_json','att_site_json','security_certificate_no',
    'level_certificate_no'
FROM source_kafka_tenant_employees c;

/**
  来自kafka的部门信息
 */
CREATE TABLE source_kafka_tenant_departments(
                                                id STRING,name STRING,code STRING,organization_id STRING,
                                               `event_time` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp',
                                               PRIMARY KEY (id) NOT ENFORCED
)WITH (
     'connector' = 'kafka',
     'topic' = 'obpm2.binlog-cdc.topic.v2.tenant_departments',
     'properties.bootstrap.servers' = '117.78.39.204:9092',
     'properties.group.id' = 'kafka-starrock-client',
     'scan.startup.mode' = 'earliest-offset',
     'format' = 'debezium-json'
     );

/**
  部门的持久化明细信息
 */
CREATE TABLE sink_dwd_tenant_departments(
    id STRING,name VARCHAR(200),code STRING,organization_id STRING,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
        'connector' = 'starrocks',
        'jdbc-url' = 'jdbc:mysql://127.0.0.1:9030',
        'load-url' = '127.0.0.1:8030',
        'database-name' = 'olap_dwd',
        'table-name' = 'dwd_tenant_departments',
        'username' = 'root',
        'password' = '',
        'sink.buffer-flush.interval-ms' = '5000',
        'sink.properties.column_separator' = '\x01',
        'sink.properties.row_delimiter' = '\x02',
        'sink.properties.partial_update'='true'
);

insert into sink_dwd_tenant_departments
(id,name,code,organization_id)
select
    c.id,c.name,c.code,c.organization_id
from source_kafka_tenant_departments c;

/**
  职员所在部门的汇总后台
 */
CREATE TABLE IF NOT EXISTS sink_dwm_employee_departments (
                                                             employee_id STRING,
                                                             department_json STRING,
                                                             last_sync_time TIMESTAMP(3),
    PRIMARY KEY (employee_id) NOT ENFORCED
    ) with (
          'connector' = 'starrocks',
          'jdbc-url' = 'jdbc:mysql://127.0.0.1:9030',
          'load-url' = '127.0.0.1:8030',
          'database-name' = 'olap_dwd',
          'table-name' = 'dwm_employee_departments',
          'username' = 'root',
          'password' = '',
          'sink.buffer-flush.interval-ms' = '5000',
          'sink.properties.column_separator' = '\x01',
          'sink.properties.row_delimiter' = '\x02',
          'sink.properties.partial_update'='true'
          );

/**
  来自binlog的信息部门人员信息表
 */
CREATE TABLE source_binlog_tenant_department_employee_relations(
     id STRING,created_time BIGINT,master_slave_type INT,department_id STRING,employee_id STRING,
     PRIMARY KEY (id) NOT ENFORCED
)
    WITH (
        'connector'='mysql-cdc',
        'hostname'='49.4.21.141',
        'port' = '45611',
        'username'='bcx',
        'password'='Wstestv5qy#2022',
        'database-name' = 'obpm2',
        'table-name'='tenant_department_employee_relations',
        'server-id'='5410-5412',
        'scan.startup.mode'='earliest-offset'
        );



CREATE TABLE source_dwd_tenant_departments(
                                            id STRING,name VARCHAR(200),code STRING,organization_id STRING,
                                            PRIMARY KEY (id) NOT ENFORCED
) WITH (
      'connector' = 'starrocks',
      'jdbc-url' = 'jdbc:mysql://127.0.0.1:9030',
      'scan-url' = '127.0.0.1:8030',
      'database-name' = 'olap_dwd',
      'table-name' = 'dwd_tenant_departments',
      'username' = 'root',
      'password' = ''
      );

insert into sink_dwm_employee_departments
(employee_id,department_json,last_sync_time)
select
    std.employee_id, sjtd.name, TO_TIMESTAMP(FROM_UNIXTIME(std.created_time/1000, 'yyyy-MM-dd HH:mm:ss'))
from source_binlog_tenant_department_employee_relations std join source_dwd_tenant_departments sjtd on std.department_id=sjtd.id;