--创建kafka表
drop table if EXISTS kafka_run_log;
create table kafka_run_log(
  carId int,
  fuel_consumption int,
  traffic string,
  speed int,
  latitude double,
  longitude double,
  create_time TIMESTAMP(3)
) with (
  'connector.type' = 'kafka',
  'connector.version' = '0.10',
  'connector.topic' = 'vehicle_run_info',
  'connector.properties.bootstrap.servers' = ' XX.XX.XX.XX:9093',
  'connector.startup-mode' = 'latest-offset',
  'connector.properties.enable.auto.commit' = 'false',
  'format.type' = 'csv'
);
/*
 创建mysql表语句
 */
drop table if EXISTS mysql_run_log;
CREATE TABLE mysql_run_log (
  carId int,
  fuel_consumption INT,
  traffic varchar(50),
  speed int,
  latitude double,
  longitude double,
  create_time TIMESTAMP(3)
) WITH (
  'connector.type' = 'jdbc',
  'connector.url' = 'jdbc:mysql://XX.XX.XX.XX:3306/car_run_log?useSSL=false',
  'connector.table' = 'mysql_run_log',
  'connector.username' = 'edap_test',
  'connector.password' = 'edap_test@1'
);
insert into
  mysql_run_log
select
  carId,
  fuel_consumption,
  traffic,
  speed,
  latitude,
  longitude,
  create_time
from
  kafka_run_log;
select
  count(carId)
from
  mysql_run_log;