基于SQL实战

案例演示：基于Flink SQL实现Flink CDC从MySQL数据库flink_db.tb_product表实时获取增量数据，打印控制台。

代码：

xxxxxxxxxx
package com.itbaizhan.flink.java.cdc;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class CdcSQLDemo {
    public static void main(String[] args) {
        //1.创建执行环境对象
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.设置checkpoint
        env.enableCheckpointing(1000, CheckpointingMode.AT_LEAST_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///E:/flink_checkpoint2");
        //3.设置并行度
        env.setParallelism(1);

        //4.构建环境配置对象，指定流式处理
        EnvironmentSettings settings = EnvironmentSettings
                .newInstance()
                .inStreamingMode()
                .build();
        //5.创建表环境对象
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);
        //6.创建输入表，使用CDC-SQL-MySQL获取数据
        tableEnv.executeSql(
                "CREATE TABLE tb_binlog (\n" +
                        " id INT NOT NULL,\n" +
                        " name STRING,\n" +
                        " price DOUBLE,\n" +
                        " PRIMARY KEY(id) NOT ENFORCED\n" +
                        ") WITH (\n" +
                        " 'connector' = 'mysql-cdc',\n" +
                        " 'hostname' = 'node1',\n" +
                        " 'port' = '3306',\n" +
                        " 'username' = 'root',\n" +
                        " 'password' = '123456',\n" +
                        " 'database-name' = 'flink_db',\n" +
                        " 'table-name' = 'tb_product',\n" +
                        " 'scan.startup.mode' = 'initial'\n" +
                        ")"
        );
        //7.查询数据
        TableResult tableResult = tableEnv.executeSql("select id,name,price from tb_binlog");
        //8.输出
        tableResult.print();
    }
}

运行程序后查看控制台。

插入数据后再次查看控制台：

INSERT INTO tb_product(id,NAME,price) VALUES(11,'macbook3',9998);

INSERT INTO tb_product(id,NAME,price) VALUES(12,'macbook4',8886);

更多案例：https://space.bilibili.com/33807709