Java应用Flink CDC监听MySQL数据变更内容输出到控制台

maven 依靠

1. <properties>
2.         <flink.version>1.14.0</flink.version>
3.         <flink-cdc.version>2.3.0</flink-cdc.version>
4.     </properties>
5.     <dependencies>
6.         <!-- Flink dependencies -->
7.         <dependency>
8.             <groupId>org.apache.flink</groupId>
9.             <artifactId>flink-connector-base</artifactId>
10.             <version>${flink.version}</version>
11.         </dependency>
12.         <dependency>
13.             <groupId>com.ververica</groupId>
14.             <artifactId>flink-sql-connector-mysql-cdc</artifactId>
15.             <version>${flink-cdc.version}</version>
16.         </dependency>
18.         <dependency>
19.             <groupId>org.apache.flink</groupId>
20.             <artifactId>flink-streaming-java_2.12</artifactId>
21.             <version>${flink.version}</version>
22.         </dependency>
24.         <dependency>
25.             <groupId>org.apache.flink</groupId>
26.             <artifactId>flink-clients_2.12</artifactId>
27.             <version>${flink.version}</version>
28.         </dependency>
30.         <dependency>
31.             <groupId>org.apache.flink</groupId>
32.             <artifactId>flink-table-api-java-bridge_2.12</artifactId>
33.             <version>${flink.version}</version>
34.         </dependency>
36.         <dependency>
37.             <groupId>org.apache.flink</groupId>
38.             <artifactId>flink-runtime-web_2.12</artifactId>
39.             <version>${flink.version}</version>
40.         </dependency>
41.     </dependencies>

复制代码

自定义数据变化处理器

1. package org.example;
3. import org.apache.flink.configuration.Configuration;
4. import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
6. public class CustomSink extends RichSinkFunction<String> {
7.     @Override
8.     public void open(Configuration parameters) throws Exception {
9.         super.open(parameters);
10.     }
12.     @Override
13.     public void close() throws Exception {
14.         super.close();
15.     }
17.     @Override
18.     public void invoke(String value, Context context) throws Exception {
19.         //0P字段，该字段也有4种取值。分别是C(Create ) , U(Updlate) . D(Delete )，Read 。
20.         // 对于U操作,其数据部分同时包含了Before和After.
21.          System.out.println(">>>" + value);
23.     }
24. }

复制代码

flink cdc监听

1. package org.example;
3. import com.ververica.cdc.connectors.mysql.source.MySqlSource;
4. import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
5. import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
6. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
7. import org.apache.flink.configuration.Configuration;
8. import org.apache.flink.configuration.RestOptions;
9. import org.apache.flink.streaming.api.datastream.DataStreamSink;
10. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
12. public class MysqlSourceExample {
13.     public static void main(String[] args) throws Exception {
14.         DebeziumDeserializationSchema debeziumDeserializationSchema = new JsonDebeziumDeserializationSchema();
15.         MySqlSource<String> source = MySqlSource.builder()
16.                 .hostname("127.0.0.1")
17.                 .port(3306)
18.                 .databaseList("canal_manager")// set captured database
19.                 .tableList("canal_manager.canal_user")// set captured table
20.                 .startupOptions(StartupOptions.latest()) // 设置从最新的修改记录开始读取
21.                 .username("root")
22.                 .password("123456")
23.                 .deserializer(debeziumDeserializationSchema) // converts SourceRecord to JSON string
24.                 .includeSchemaChanges(true)
25.                 .build();
26.         //启动一个webuI。
27.         Configuration configuration = new Configuration();
28.         configuration.setInteger(RestOptions.PORT, 8081);
29.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
31.         //检者点间隔时间
32.         env .enableCheckpointing(5000);
33.         DataStreamSink<String> sink = env.fromSource(source, WatermarkStrategy.noWatermarks(), "MySQL Source")
34.                 .addSink(new CustomSink());
35.         env.execute();
37.     }
38. }

复制代码

验证

启动后web页面地址访问http://localhost:8081/，MySQL数据库canal_manager中的表canal_user数据发生修改，控制台有输出json：

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

继续阅读请点击广告