Flink Table 将Stream直接写入MySQL数据库

Dear 丶 2021-12-05 14:37 257阅读 0赞

## Flink Table 将Stream直接写入MySQL数据库 ##

Flink Table提供了一个`JDBCAppendTableSink`，可以直接往可靠地数据库中Sink数据，下面以MySQL为例：

添加Maven的pom.xml依赖

<dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-jdbc_2.11</artifactId>
        <version>1.8.0</version>
    </dependency>

代码如下：

public class SqlSinkJdbcStream { 
        public static void main(String[] args) throws Exception { 
    
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
            // 必须设置checkpoint的间隔时间，不然不会写入jdbc
            env.enableCheckpointing(5000L);
            StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
    
            Schema schema = new Schema()
                    .field("userId", Types.STRING)
                    .field("name", Types.STRING)
                    .field("age", Types.STRING)
                    .field("sex", Types.STRING)
                    .field("createTime", Types.BIG_DEC)
                    .field("updateTime", Types.BIG_DEC);
    
            tableEnv
                    .connect(
                            new Kafka().version("0.10").topic("user").property("bootstrap.servers", "localhost:9092")
                    )
                    .withSchema(schema)
                    .withFormat(new Json().deriveSchema())
                    .inAppendMode()
                    .registerTableSource("Users");
    
            Table table = tableEnv.sqlQuery("select userId,name,age,sex,createTime,updateTime from Users");
            DataStream<Row> result = tableEnv.toAppendStream(table, TypeInformation.of(Row.class));
            result.print();
    
            JDBCAppendTableSink sink = new JDBCAppendTableSinkBuilder()
                    .setDBUrl("jdbc:mysql://localhost:3306/test?useSSL=false")
                    .setDrivername("com.mysql.jdbc.Driver")
                    .setUsername("root")
                    .setPassword("root")
                    .setBatchSize(1000)
                    .setQuery("REPLACE INTO user(userId,name,age,sex,createTime,updateTime) values(?,?,?,?,?,?)")
                    .setParameterTypes(new TypeInformation[]{ Types.STRING, Types.STRING, Types.STRING, Types.STRING, Types.BIG_DEC, Types.BIG_DEC})
                    .build();
    
            tableEnv.registerTableSink("Result",
                    new String[]{ "userId", "name", "age", "sex", "createTime", "updateTime"},
                    new TypeInformation[]{ Types.STRING, Types.STRING, Types.STRING, Types.STRING, Types.BIG_DEC, Types.BIG_DEC},
                    sink);
    
            tableEnv.insertInto(table, "Result", new StreamQueryConfig());
    
            env.execute("SqlSinkJdbcStream");
        }
    }
    // mysql的建表语句
    /* CREATE TABLE `user` ( `userId` varchar(10) NOT NULL, `name` varchar(10) DEFAULT NULL, `age` varchar(3) DEFAULT NULL, `sex` varchar(10) DEFAULT NULL, `createTime` varchar(20) DEFAULT NULL, `updateTime` varchar(20) DEFAULT NULL, PRIMARY KEY (`userId`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8; */

注意：需要设置checkpoint的true，设置checkpoint间隔时间。