在Flink1.9.x版本中,社区版本的 Flink 新增 了一个 SQL DDL 的新特性，但是暂时还不支持流式的一些概念的定义,比如说水位(watermark). 下面主要介绍一下怎么使用DDL创建kafak源表.
定义create table语句从kafka中读取数据
""" |CREATE TABLE PERSON ( | name VARCHAR COMMENT '姓名', | age VARCHAR COMMENT '年龄', | city VARCHAR COMMENT '所在城市', | address VARCHAR COMMENT '家庭住址', | ts TIMESTAMP COMMENT '时间戳' |) |WITH ( | 'connector.type' = 'kafka', -- 使用 kafka connector | 'connector.version' = '0.11', -- kafka 版本 | 'connector.topic' = 'xxx', -- kafka topic | 'connector.startup-mode' = 'latest-offset', -- 从最新的 offset 开始读取 | 'connector.properties.0.key' = 'zookeeper.connect', -- 连接信息 | 'connector.properties.0.value' = 'xxx', | 'connector.properties.1.key' = 'bootstrap.servers', | 'connector.properties.1.value' = 'xxx', | 'update-mode' = 'append', | 'format.type' = 'json', -- 数据源格式为 json | 'format.derive-schema' = 'true' -- 从 DDL schema 确定 json 解析规则 |)
如上面的 sql，基本语法是 create table () with ()
with 后面是一些基本的属性，比如 connector.type 描述了 从 kafka 中读取数据
connector.version 描述了 使用的是哪个版本的 kafka
connector.topic 描述了 从 哪个 topic 中读取数据
connector.startup-mode 描述了 从 哪个位置开始读取数据 等等。
可能有同学会觉得其中的 connector.properties.0.key 等参数比较奇怪，社区计划将在下一个版本中改进并简化 connector 的参数配置。
需要注意的是:
1,表的字段要区分大小写
2,字段里面不能有timestamp ,watermark等关键字
完整的代码如下
package flink.ddl import org.apache.flink.api.scala._ import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment import org.apache.flink.table.api.EnvironmentSettings import org.apache.flink.table.api.scala._ import org.apache.flink.types.Row /** * @program: Flink1.9.0 * @description: ${description} * @author: JasonLee * @create: 2020-01-14 19:49 */ object FlinkKafkaDDLDemo { def main(args: Array[String]): Unit = { val env = StreamExecutionEnvironment.getExecutionEnvironment env.setParallelism(3) val settings = EnvironmentSettings.newInstance() .useBlinkPlanner() .inStreamingMode() .build() val tEnv = StreamTableEnvironment.create(env, settings) val createTable = """ |CREATE TABLE PERSON ( | name VARCHAR COMMENT '姓名', | age VARCHAR COMMENT '年龄', | city VARCHAR COMMENT '所在城市', | address VARCHAR COMMENT '家庭住址', | ts TIMESTAMP COMMENT '时间戳' |) |WITH ( | 'connector.type' = 'kafka', -- 使用 kafka connector | 'connector.version' = '0.11', -- kafka 版本 | 'connector.topic' = 'xxx', -- kafka topic | 'connector.startup-mode' = 'latest-offset', -- 从最新的 offset 开始读取 | 'connector.properties.0.key' = 'zookeeper.connect', -- 连接信息 | 'connector.properties.0.value' = 'xxx', | 'connector.properties.1.key' = 'bootstrap.servers', | 'connector.properties.1.value' = 'xxx', | 'update-mode' = 'append', | 'format.type' = 'json', -- 数据源格式为 json | 'format.derive-schema' = 'true' -- 从 DDL schema 确定 json 解析规则 |) """.stripMargin tEnv.sqlUpdate(createTable) val query = """ |SELECT name,COUNT(age) FROM PERSON GROUP BY name """.stripMargin val result = tEnv.sqlQuery(query) result.toRetractStream[Row].print() tEnv.execute("Flink SQL DDL") } }
我们把上面的代码提交到集群,可以看到任务的DAG图,如下图所示:
然后我们向上面配置的topic写入数据,可以看下Flink的UI输出的结果:
可以看到这是上面那个SQL打印的结果.也可以通过DDL直接把结果写入mysql库中,今天就主要介绍到这里,后面会带来更多相关的内容.
