Flink tableapi数据写入MySQL

2021/10/19 19:14:20

本文主要是介绍Flink tableapi数据写入MySQL,对大家解决编程问题具有一定的参考价值,需要的程序猿们随着小编来一起学习吧!

jdbc 连接的代码实现比较特殊,因为没有对应的 java/scala 类实现 ConnectorDescriptor,所以不能直接 tableEnv.connect()。不过 Flink SQL留下了执行 DDL的接口:tableEnv.sqlUpdate()。
对于 jdbc 的创建表操作,天生就适合直接写 DDL 来实现,所以我们的代码可以这样写:

val sinkDDL: String =
"""
|create table jdbcOutputTable (
| id varchar(20) not null,
| cnt bigint not null
|) with (
| 'connector.type' = 'jdbc',
| 'connector.url' = 'jdbc:mysql://localhost:3306/test',
| 'connector.table' = 'sensor_count',
| 'connector.driver' = 'com.mysql.jdbc.Driver',
| 'connector.username' = 'root',
| 'connector.password' = '123456'
|)
""".stripMargin
tableEnv.sqlUpdate(sinkDDL)
aggResultSqlTable.insertInto("jdbcOutputTable")



这篇关于Flink tableapi数据写入MySQL的文章就介绍到这儿,希望我们推荐的文章对大家有所帮助,也希望大家多多支持为之网!


扫一扫关注最新编程教程