从kafka读取3个topic,join后写数据到clickhouse

代码示例：

val createTable =
  """
    |CREATE TABLE nt_sale_order (
    |    id VARCHAR,
    |    write_date BIGINT,
    |    create_uid INT,
    |    name VARCHAR,
    |    op VARCHAR
    |)
    |WITH (
    |    'connector' = 'kafka',
    |    'topic' = 'shopforce.public.nt_sale_order',
    |    'scan.startup.mode' = 'latest-offset',
    |    'properties.bootstrap.servers' = '192.168.10.16:9092',
    |    'properties.group.id' = 'testGroup',
    |    'format' = 'debezium-json',
    |    'debezium-json.schema-include' = 'true'
    |)
  """.stripMargin

val createTable1 =
  """
    |CREATE TABLE nt_sale_order_line (
    |    id VARCHAR,
    |    order_id VARCHAR,
    |    unit_price VARCHAR,
    |    op VARCHAR
    |)
    |WITH (
    |    'connector' = 'kafka',
    |    'topic' = 'shopforce.public.nt_sale_order_line',
    |    'scan.startup.mode' = 'latest-offset',
    |    'properties.bootstrap.servers' = '192.168.10.16:9092',
    |    'properties.group.id' = 'testGroup1',
    |    'format' = 'debezium-json',
    |    'debezium-json.schema-include' = 'true'
    |)
  """.stripMargin


val createTable2 =
  """
    |CREATE TABLE nt_order_payment (
    |    id VARCHAR,
    |    order_id VARCHAR,
    |    name VARCHAR,
    |    paid_at VARCHAR,
    |    op VARCHAR
    |)
    |WITH (
    |    'connector' = 'kafka',
    |    'topic' = 'shopforce.public.nt_payment',
    |    'scan.startup.mode' = 'latest-offset',
    |    'properties.bootstrap.servers' = '192.168.10.16:9092',
    |    'properties.group.id' = 'testGroup1',
    |    'format' = 'debezium-json',
    |    'debezium-json.schema-include' = 'true'
    |)
  """.stripMargin

tblEnv.executeSql(createTable)
tblEnv.executeSql(createTable1)
tblEnv.executeSql(createTable2)


val nt_order_detail =
  """
    |CREATE TABLE nt_order_detail(
    |id VARCHAR primary key,
    |name VARCHAR,
    |unit_price VARCHAR,
    |paid_at VARCHAR
    |)
    |WITH(
    |'connector' = 'jdbc',
    |'url' = 'jdbc:clickhouse://192.168.10.16:8123/default',
    |'table-name' = 'nt_order_detail'
    |)
  """.stripMargin

tblEnv.executeSql(nt_order_detail)


var sqlQuery =
  """
    |insert into nt_order_detail
    |SELECT id,name,unit_price,paid_at from(
    |SELECT p.id,b.name,p.unit_price,s.paid_at from nt_sale_order_line AS p
    	|JOIN nt_sale_order AS b
    |  ON b.id = p.order_id
    	|JOIN nt_order_payment AS s
    |  ON b.id = s.order_id where b.op = 'c' and p.op = 'c' and s.paid_at is not null)a
    |""".stripMargin

tblEnv.executeSql(sqlQuery)

执行：