目录

1.写在前面

2.Maven依赖

3.代码实现-普通实现

4.集群测试

4.1 环境准备

4.2 查看任务结果

1.写在前面

        Flink CDC有两种实现方式&＃xff0c;一种是DataStream&＃xff0c;另一种是FlinkSQL方式。

• DataStream方式&＃xff1a;优点是可以应用于多库多表&＃xff0c;缺点是需要自定义反序列化器&＃xff08;灵活&＃xff09;
• FlinkSQL方式&＃xff1a;优点是不需要自定义反序列化器&＃xff0c;缺点是只能应用于单表查询

2.Maven依赖

org.apache.flinkflink-java1.12.7org.apache.flinkflink-streaming-java_2.121.12.7org.apache.flinkflink-clients_2.121.12.7org.apache.hadoophadoop-client2.7.7mysqlmysql-connector-java5.1.49com.alibaba.ververicaflink-connector-mysql-cdc1.2.0com.alibabafastjson1.2.75org.apache.flinkflink-table-planner-blink_2.121.12.7org.apache.maven.pluginsmaven-assembly-plugin3.0.0jar-with-dependenciesmake-assemblypackagesingleorg.apache.maven.pluginsmaven-compiler-plugin88

3.代码实现-普通实现

package com.atguigu;import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;public class FlinkCDCWithSQL {public static void main(String[] args) throws Exception {//1.获取执行环境StreamExecutionEnvironment env &＃61; StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);StreamTableEnvironment tableEnv &＃61; StreamTableEnvironment.create(env);//2.DDL方式建表,flink_sql的方式只能监控一张表tableEnv.executeSql("CREATE TABLE mysql_binlog ( " &＃43;" id STRING NOT NULL, " &＃43;" tm_name STRING, " &＃43;" logo_url STRING " &＃43;") WITH ( " &＃43;" &＃39;connector&＃39; &＃61; &＃39;mysql-cdc&＃39;, " &＃43;" &＃39;hostname&＃39; &＃61; &＃39;192.168.0.111&＃39;, " &＃43;" &＃39;port&＃39; &＃61; &＃39;3306&＃39;, " &＃43;" &＃39;username&＃39; &＃61; &＃39;root&＃39;, " &＃43;" &＃39;password&＃39; &＃61; &＃39;123456&＃39;, " &＃43;" &＃39;database-name&＃39; &＃61; &＃39;gmall2021&＃39;, " &＃43;" &＃39;table-name&＃39; &＃61; &＃39;base_trademark&＃39; " &＃43;")");//3.查询数据Table table &＃61; tableEnv.sqlQuery("select * from mysql_binlog");//4.将动态表转换为流DataStream> retractStream &＃61; tableEnv.toRetractStream(table, Row.class);retractStream.print();//5.启动任务env.execute("FlinkCDCWithSQL");}}


4.集群测试

4.1 环境准备

1. 启动ha-hadoop集群&＃xff1a;sh ha-hadoop.sh start

2. 创建作业归档目录&＃xff0c;只需要创建一次&＃xff1a;hdfs dfs -mkdir /flink-jobhistory

3. 启动Flink集群和任务历史服务

   1. start-cluster.sh
   2. historyserver.sh start
4. 运行该Flink任务&＃xff1a;/opt/software/flink-1.12.7/bin/flink run -m yarn-cluster -ys 1 -ynm gmall-flink-cdc -c com.ucas.FlinkCDCWithCustomerDeserialization -d /root/mySoftware/gmall-flink-cdc.jar

4.2 查看任务结果

&＃xff08;1&＃xff09;打开yarn&＃xff0c;查看任务&＃xff1a;http://192.168.0.112:8088/cluster/apps&＃xff0c;并且通过id点击进去

&＃xff08;2&＃xff09;点击Tracking URL&＃xff0c;进入FlinkWeb界面

 &＃xff08;3&＃xff09; 打开左侧TaskManagers中的Stdout查看控制台输出信息