1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
| StreamExecutionEnvironment ssEnv &＃061; StreamExecutionEnvironment.getExecutionEnvironment();
EnvironmentSettings ssSettings &＃061; EnvironmentSettings.newInstance().useOldPlanner().inStreamingMode().build();
StreamTableEnvironment ssTableEnv &＃061; StreamTableEnvironment.create(ssEnv, ssSettings);
String sql &＃061; "create table test_table ( \n" &＃043;
"id int,\n" &＃043;
"fk_drug_info int,\n" &＃043;
"fk_professional_id string,\n" &＃043;
"urine_time TIMESTAMP,\n" &＃043;
"urine_place string,\n" &＃043;
"urine_result int,\n" &＃043;
"test_type int,\n" &＃043;
"test_time TIMESTAMP,\n" &＃043;
"seal_time TIMESTAMP,\n" &＃043;
"urine_image string,\n" &＃043;
"urine_video string,\n" &＃043;
"urine_cause string,\n" &＃043;
"help_professional_id int,\n" &＃043;
"latitude string,\n" &＃043;
"fk_auth_manager int,\n" &＃043;
"table_date int,\n" &＃043;
"urine_state_type int,\n" &＃043;
"annex string,\n" &＃043;
"update_time int,\n" &＃043;
"is_del int\n"&＃043;
") with ( \n" &＃043;
"&＃039;connector&＃039; &＃061; &＃039;kafka&＃039;,\n" &＃043;
"&＃039;topic&＃039; &＃061; &＃039;LG831_test.dbo.urine_upload&＃039;,\n" &＃043;
"&＃039;properties.bootstrap.servers&＃039; &＃061; &＃039;hadoop-node1:9092,hadoop-node2:9092,hadoop-node3:9092&＃039;,\n" &＃043;
"&＃039;properties.group.id&＃039; &＃061; &＃039;connect-cluster&＃039;,\n" &＃043;
"&＃039;format&＃039; &＃061; &＃039;debezium-json&＃039;"&＃043;
")";
ssTableEnv.executeSql(sql);
Table result &＃061; ssTableEnv.sqlQuery("select * from test_table");
result.execute().print(); |
京公网安备 11010802041100号