Flink 1.10 SQL、HiveCatalog与事件时间整合示例

本文已授权「Flink中文社区」微信公众号发布并标注原创。

前言

Flink 1.10与1.9相比又是个船新版本,在我们感兴趣的很多方面都有改进,特别是Flink SQL。本文用根据埋点日志计算PV、UV的简单示例来体验Flink 1.10的两个重要新特性,一是SQL DDL对事件时间的支持,二是Hive Metastore作为Flink的元数据存储(即HiveCatalog)。这两点将会为我们构建实时数仓提供很大的便利。

添加依赖项

示例采用Hive版本为1.1.0,Kafka版本为0.11.0.2。

要使Flink与Hive集成以使用HiveCatalog,需要先将以下JAR包放在${FLINK_HOME}/lib目录下。

  • flink-connector-hive_2.11-1.10.0.jar
  • flink-shaded-hadoop-2-uber-2.6.5-8.0.jar
  • hive-metastore-1.1.0.jar
  • hive-exec-1.1.0.jar
  • libfb303-0.9.2.jar

后三个JAR包都是Hive自带的,可以在${HIVE_HOME}/lib目录下找到。前两个可以通过阿里云Maven搜索GAV找到并手动下载(groupId都是org.apache.flink)。

再在pom.xml内添加相关的Maven依赖。

  <properties>
    <scala.bin.version>2.11</scala.bin.version>
    <flink.version>1.10.0</flink.version>
    <hive.version>1.1.0</hive.version>
  </properties>

  <dependencies>
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-table-api-scala_${scala.bin.version}</artifactId>
      <version>${flink.version}</version>
    </dependency>
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-table-api-scala-bridge_${scala.bin.version}</artifactId>
      <version>${flink.version}</version>
    </dependency>
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-table-planner-blink_${scala.bin.version}</artifactId>
      <version>${flink.version}</version>
    </dependency>
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-sql-connector-kafka-0.11_${scala.bin.version}</artifactId>
      <version>${flink.version}</version>
    </dependency>
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-connector-hive_${scala.bin.version}</artifactId>
      <version>${flink.version}</version>
    </dependency>
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-json</artifactId>
      <version>${flink.version}</version>
    </dependency>
    <dependency>
      <groupId>org.apache.hive</groupId>
      <artifactId>hive-exec</artifactId>
      <version>${hive.version}</version>
    </dependency>
  </dependencies>

最后,找到Hive的配置文件hive-site.xml并加入项目,准备工作就完成了。

注册HiveCatalog、创建数据库

不多废话了,直接上代码,简洁易懂。

    val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment
    streamEnv.setParallelism(5)
    streamEnv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val tableEnvSettings = EnvironmentSettings.newInstance()
        .useBlinkPlanner()
        .inStreamingMode()
        .build()
    val tableEnv = StreamTableEnvironment.create(streamEnv, tableEnvSettings)

    val catalog = new HiveCatalog(
      "rtdw",                   // catalog name
      "default",                // default database
      "/Users/lmagic/develop",  // Hive config (hive-site.xml) directory
      "1.1.0"                   // Hive version
    )
    tableEnv.registerCatalog("rtdw", catalog)
    tableEnv.useCatalog("rtdw")

    val createDbSql = "CREATE DATABASE IF NOT EXISTS rtdw.ods"
    tableEnv.sqlUpdate(createDbSql)

创建Kafka流表并指定事件时间

我们的埋点日志存储在指定的Kafka topic里,为JSON格式,简化版schema大致如下。

{
    "eventType": "clickBuyNow",
    "userId": "97470180",
    "shareUserId": "",
    "platform": "xyz",
    "columnType": "merchDetail",
    "merchandiseId": "12727495",
    "fromType": "wxapp",
    "siteId": "20392",
    "categoryId": "",
    "ts": 1585136092541
}

其中ts字段就是埋点事件的时间戳(毫秒)。在Flink 1.9时代,用CREATE TABLE语句创建流表时是无法指定事件时间的,只能默认用处理时间。而在Flink 1.10下,可以这样写。

CREATE TABLE rtdw.ods.streaming_user_active_log (
  eventType STRING COMMENT '...',
  userId STRING,
  shareUserId STRING,
  platform STRING,
  columnType STRING,
  merchandiseId STRING,
  fromType STRING,
  siteId STRING,
  categoryId STRING,
  ts BIGINT,
  procTime AS PROCTIME(), -- 处理时间
  eventTime AS TO_TIMESTAMP(FROM_UNIXTIME(ts / 1000, 'yyyy-MM-dd HH:mm:ss')), -- 事件时间
  WATERMARK FOR eventTime AS eventTime - INTERVAL '10' SECOND -- 水印
) WITH (
  'connector.type' = 'kafka',
  'connector.version' = '0.11',
  'connector.topic' = 'ng_log_par_extracted',
  'connector.startup-mode' = 'latest-offset', -- 指定起始offset位置
  'connector.properties.zookeeper.connect' = 'zk109:2181,zk110:2181,zk111:2181',
  'connector.properties.bootstrap.servers' = 'kafka112:9092,kafka113:9092,kafka114:9092',
  'connector.properties.group.id' = 'rtdw_group_test_1',
  'format.type' = 'json',
  'format.derive-schema' = 'true', -- 由表schema自动推导解析JSON
  'update-mode' = 'append'
)

Flink SQL引入了计算列(computed column)的概念,其语法为column_name AS computed_column_expression,它的作用是在表中产生数据源schema不存在的列,并且可以利用原有的列、各种运算符及内置函数。比如在以上SQL语句中,就利用内置的PROCTIME()函数生成了处理时间列,并利用原有的ts字段与FROM_UNIXTIME()、TO_TIMESTAMP()两个时间转换函数生成了事件时间列。

为什么ts字段不能直接用作事件时间呢?因为Flink SQL规定时间特征必须是TIMESTAMP(3)类型,即形如"yyyy-MM-ddTHH:mm:ssZ"格式的字符串,Unix时间戳自然是不行的,所以要先转换一波。

既然有了事件时间,那么自然要有水印。Flink SQL引入了WATERMARK FOR rowtime_column_name AS watermark_strategy_expression的语法来产生水印,有以下两种通用的做法:

  • 单调不减水印(对应DataStream API的AscendingTimestampExtractor)
WATERMARK FOR rowtime_column AS rowtime_column - INTERVAL '0.001' SECOND
  • 有界乱序水印(对应DataStream API的BoundedOutOfOrdernessTimestampExtractor)
WATERMARK FOR rowtime_column AS rowtime_column - INTERVAL 'n' TIME_UNIT

上文的SQL语句中就是设定了10秒的乱序区间。如果看官对水印、AscendingTimestampExtractor和BoundedOutOfOrdernessTimestampExtractor不熟的话,可以参见之前的这篇,就能理解为什么会是这样的语法了。

下面来正式建表。

    val createTableSql =
      """
        |上文的SQL语句
        |......
      """.stripMargin
    tableEnv.sqlUpdate(createTableSql)

执行完毕后,我们还可以去到Hive执行DESCRIBE FORMATTED ods.streaming_user_active_log语句,能够发现该表并没有事实上的列,而所有属性(包括schema、connector、format等等)都作为元数据记录在了Hive Metastore中。

Flink SQL创建的表都会带有一个标记属性is_generic=true,图中未示出。

开窗计算PV、UV

用30秒的滚动窗口,按事件类型来分组,查询语句如下。

SELECT eventType,
TUMBLE_START(eventTime, INTERVAL '30' SECOND) AS windowStart,
TUMBLE_END(eventTime, INTERVAL '30' SECOND) AS windowEnd,
COUNT(userId) AS pv,
COUNT(DISTINCT userId) AS uv
FROM rtdw.ods.streaming_user_active_log
WHERE platform = 'xyz'
GROUP BY eventType, TUMBLE(eventTime, INTERVAL '30' SECOND)

关于窗口在SQL里的表达方式请参见官方文档。1.10版本SQL的官方文档写的还是比较可以的。

懒得再输出到一个结果表了,直接转换成流打到屏幕上。

    val queryActiveSql =
      """
        |......
        |......
      """.stripMargin
    val result = tableEnv.sqlQuery(queryActiveSql)

    result
        .toAppendStream[Row]
        .print()
        .setParallelism(1)

敏感数据较多,就不截图了。民那晚安。

最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 203,772评论 6 477
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 85,458评论 2 381
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 150,610评论 0 337
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 54,640评论 1 276
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 63,657评论 5 365
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 48,590评论 1 281
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 37,962评论 3 395
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 36,631评论 0 258
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 40,870评论 1 297
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,611评论 2 321
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,704评论 1 329
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,386评论 4 319
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 38,969评论 3 307
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,944评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,179评论 1 260
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 44,742评论 2 349
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 42,440评论 2 342

推荐阅读更多精彩内容