1--创建kafka表
2drop table if EXISTS kafka_run_log;
3create table kafka_run_log(
4  carId int,
5  fuel_consumption int,
6  traffic string,
7  speed int,
8  latitude double,
9  longitude double,
10  create_time TIMESTAMP(3)
11) with (
12  'connector.type' = 'kafka',
13  'connector.version' = '0.10',
14  'connector.topic' = 'vehicle_run_info',
15  'connector.properties.bootstrap.servers' = ' XX.XX.XX.XX:9093',
16  'connector.startup-mode' = 'latest-offset',
17  'connector.properties.enable.auto.commit' = 'false',
18  'format.type' = 'csv'
19);
20/*
21 创建mysql表语句
22 */
23drop table if EXISTS mysql_run_log;
24CREATE TABLE mysql_run_log (
25  carId int,
26  fuel_consumption INT,
27  traffic varchar(50),
28  speed int,
29  latitude double,
30  longitude double,
31  create_time TIMESTAMP(3)
32) WITH (
33  'connector.type' = 'jdbc',
34  'connector.url' = 'jdbc:mysql://XX.XX.XX.XX:3306/car_run_log?useSSL=false',
35  'connector.table' = 'mysql_run_log',
36  'connector.username' = 'edap_test',
37  'connector.password' = 'edap_test@1'
38);
39insert into
40  mysql_run_log
41select
42  carId,
43  fuel_consumption,
44  traffic,
45  speed,
46  latitude,
47  longitude,
48  create_time
49from
50  kafka_run_log;
51select
52  count(carId)
53from
54  mysql_run_log;