通过flinkSql将kafka和mysql连接

kafkaToKafka

1. {"user_id": "1", "page_id":"1", "status": "success"}
2. {"user_id": "1", "page_id":"1", "status": "success"}
3. {"user_id": "1", "page_id":"1", "status": "success"}
4. {"user_id": "1", "page_id":"1", "status": "success"}
5. {"user_id": "1", "page_id":"1", "status": "fail"}

复制代码

1. package com.bigdata.day07;
3. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
4. import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
6. public class _08_kafka_to_kafka {
7.     public static void main(String[] args) {
8.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
9.         StreamTableEnvironment tabEnv = StreamTableEnvironment.create(env);
10.         tabEnv.executeSql("CREATE TABLE table1 (\n" +
11.                 "  `user_id` int,\n" +
12.                 "  `page_id` int,\n" +
13.                 "  `status` STRING\n" +
14.                 ") WITH (\n" +
15.                 "  'connector' = 'kafka',\n" +
16.                 "  'topic' = 'topicA',\n" +
17.                 "  'properties.bootstrap.servers' = 'bigdata01:9092',\n" +
18.                 "  'properties.group.id' = 'g1',\n" +
19.                 "  'scan.startup.mode' = 'latest-offset',\n" +
20.                 // 这个是需要flink-json的
21.                 "  'format' = 'json'\n" +
22.                 ")");
23.         tabEnv.executeSql("CREATE TABLE table2 (\n" +
24.                 "  `user_id` int,\n" +
25.                 "  `page_id` int,\n" +
26.                 "  `status` STRING\n" +
27.                 ") WITH (\n" +
28.                 "  'connector' = 'kafka',\n" +
29.                 "  'topic' = 'topicB',\n" +
30.                 "  'properties.bootstrap.servers' = 'bigdata01:9092',\n" +
31.                 "  'properties.group.id' = 'g1',\n" +
32.                 "  'scan.startup.mode' = 'latest-offset',\n" +
33.                 "  'format' = 'json'\n" +
34.                 ")");
36.         tabEnv.executeSql("insert into table2 select * from table1 where status = 'success'");
38.     }
39. }
40. // 非常简单的代码
42. 使用executeSql后，就可以不使用execute了
43. 但是若有一个print ，那么还需要execute

复制代码

kafkaToMysql

1. 需要先在mysql中建表
2. create table t_success
3. (
4.     id      int auto_increment,
5.     user_id int         null,
6.     page_id int         null,
7.     status  varchar(20) null,
8.     constraint t_success_pk
9.         primary key (id)
10. );

复制代码

1. package com.bigdata.day07;
3. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
4. import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
6. public class _09_kafka_to_mysql {
7.     public static void main(String[] args) throws Exception {
8.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
9.         StreamTableEnvironment tabEnv = StreamTableEnvironment.create(env);
10.         tabEnv.executeSql("CREATE TABLE table1 (\n" +
11.                 "  `user_id` int,\n" +
12.                 "  `page_id` int,\n" +
13.                 "  `status` STRING\n" +
14.                 ") WITH (\n" +
15.                 "  'connector' = 'kafka',\n" +
16.                 "  'topic' = 'topicA',\n" +
17.                 "  'properties.bootstrap.servers' = 'bigdata01:9092',\n" +
18.                 "  'properties.group.id' = 'g1',\n" +
19.                 "  'scan.startup.mode' = 'latest-offset',\n" +
20.                 "  'format' = 'json'\n" +
21.                 ")");
22.         tabEnv.executeSql("CREATE TABLE table2 (\n" +
23.                 "  `user_id` int,\n" +
24.                 "  `page_id` int,\n" +
25.                 "  `status` STRING\n" +
26.                 ") WITH (\n" +
27.                 "    'connector' = 'jdbc',\n" +
28.                 "    'url' = 'jdbc:mysql://localhost:3306/edu?useUnicode=true&characterEncoding=utf8',\n" +
29.                 "    'table-name' = 't_success', \n" +
30.                 "    'username' = 'root',\n" +
31.                 "    'password' = 'root'\n" +
32.                 ")");
34.         tabEnv.executeSql("insert into table2 select * from table1 where status = 'success'");
37.     }
38. }

复制代码

readMysql

1. package com.bigdata.day07;
3. import org.apache.flink.streaming.api.datastream.DataStream;
4. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
5. import org.apache.flink.table.api.Table;
6. import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
7. import org.apache.flink.types.Row;
9. public class _10_read_mysql {
10.     public static void main(String[] args) throws Exception {
11.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
12.         StreamTableEnvironment tabEnv = StreamTableEnvironment.create(env);
13.         tabEnv.executeSql("CREATE TABLE table1 (\n" +
14.                 "  `user_id` int,\n" +
15.                 "  `page_id` int,\n" +
16.                 "  `status` STRING\n" +
17.                 ") WITH (\n" +
18.                 "    'connector' = 'jdbc',\n" +
19.                 "    'url' = 'jdbc:mysql://localhost:3306/edu?useUnicode=true&characterEncoding=utf8',\n" +
20.                 "    'table-name' = 't_success', \n" +
21.                 "    'username' = 'root',\n" +
22.                 "    'password' = 'root'\n" +
23.                 ")");
25.         Table table = tabEnv.sqlQuery("select * from table1");
27.         DataStream<Row> appendStream = tabEnv.toAppendStream(table, Row.class);
29.         appendStream.print();
30.         env.execute();
31.     }
32. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。