Flink1.10中SQL、HiveCatalog與事件時(shí)間整合的示例分析

這篇文章將為大家詳細(xì)講解有關(guān)Flink 1.10中SQL、HiveCatalog與事件時(shí)間整合的示例分析，小編覺(jué)得挺實(shí)用的，因此分享給大家做個(gè)參考，希望大家閱讀完這篇文章后可以有所收獲。

渦陽(yáng)網(wǎng)站建設(shè)公司創(chuàng)新互聯(lián),渦陽(yáng)網(wǎng)站設(shè)計(jì)制作，有大型網(wǎng)站制作公司豐富經(jīng)驗(yàn)。已為渦陽(yáng)超過(guò)千家提供企業(yè)網(wǎng)站建設(shè)服務(wù)。企業(yè)網(wǎng)站搭建\外貿(mào)網(wǎng)站建設(shè)要多少錢，請(qǐng)找那個(gè)售后服務(wù)好的渦陽(yáng)做網(wǎng)站的公司定做！

Flink 1.10 與 1.9 相比又是個(gè)創(chuàng)新版本，在我們感興趣的很多方面都有改進(jìn)，特別是 Flink SQL。本文用根據(jù)埋點(diǎn)日志計(jì)算 PV、UV 的簡(jiǎn)單示例來(lái)體驗(yàn) Flink 1.10 的兩個(gè)重要新特性：

一是 SQL DDL 對(duì)事件時(shí)間的支持；
二是 Hive Metastore 作為 Flink 的元數(shù)據(jù)存儲(chǔ)（即 HiveCatalog）。

這兩點(diǎn)將會(huì)為我們構(gòu)建實(shí)時(shí)數(shù)倉(cāng)提供很大的便利。

添加依賴項(xiàng)

示例采用 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

后三個(gè) JAR 包都是 Hive 自帶的，可以在 ${HIVE_HOME}/lib 目錄下找到。前兩個(gè)可以通過(guò) 阿里云 Maven 搜索 GAV 找到并手動(dòng)下載（groupId 都是org.apache.flink）。

再在 pom.xml 內(nèi)添加相關(guān)的 Maven 依賴。

Maven 下載：
https://maven.aliyun.com/mvn/search

   
     
   
   
   <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，準(zhǔn)備工作就完成了。

注冊(cè) HiveCatalog、創(chuàng)建數(shù)據(jù)庫(kù)

不多廢話了，直接上代碼，簡(jiǎn)潔易懂。

   
     
   
   
   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)

創(chuàng)建 Kafka 流表并指定事件時(shí)間

我們的埋點(diǎn)日志存儲(chǔ)在指定的 Kafka topic 里，為 JSON 格式，簡(jiǎn)化版 schema 大致如下。

   
     
   
   
   

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

其中 ts 字段就是埋點(diǎn)事件的時(shí)間戳（毫秒）。在 Flink 1.9 時(shí)代，用 CREATE TABLE 語(yǔ)句創(chuàng)建流表時(shí)是無(wú)法指定事件時(shí)間的，只能默認(rèn)用處理時(shí)間。而在 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(), -- 處理時(shí)間  eventTime AS TO_TIMESTAMP(FROM_UNIXTIME(ts / 1000, 'yyyy-MM-dd HH:mm:ss')), -- 事件時(shí)間  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', -- 指定起始o(jì)ffset位置  '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自動(dòng)推導(dǎo)解析JSON  'update-mode' = 'append')

Flink SQL 引入了計(jì)算列（computed column）的概念，其語(yǔ)法為 column_name AS computed_column_expression，它的作用是在表中產(chǎn)生數(shù)據(jù)源 schema 不存在的列，并且可以利用原有的列、各種運(yùn)算符及內(nèi)置函數(shù)。比如在以上 SQL 語(yǔ)句中，就利用內(nèi)置的 PROCTIME() 函數(shù)生成了處理時(shí)間列，并利用原有的 ts 字段與 FROM_UNIXTIME()、TO_TIMESTAMP() 兩個(gè)時(shí)間轉(zhuǎn)換函數(shù)生成了事件時(shí)間列。

為什么 ts 字段不能直接用作事件時(shí)間呢？因?yàn)?Flink SQL 規(guī)定時(shí)間特征必須是 TIMESTAMP(3) 類型，即形如"yyyy-MM-ddTHH:mm:ssZ"格式的字符串，Unix 時(shí)間戳自然是不行的，所以要先轉(zhuǎn)換一波。

既然有了事件時(shí)間，那么自然要有水印。Flink SQL 引入了 WATERMARK FOR rowtime_column_name AS watermark_strategy_expression 的語(yǔ)法來(lái)產(chǎn)生水印，有以下兩種通用的做法：

單調(diào)不減水?。▽?duì)應(yīng) DataStream API 的 AscendingTimestampExtractor）

   
     
   
   
   WATERMARK FOR rowtime_column AS rowtime_column - INTERVAL '0.001' SECOND

有界亂序水印（對(duì)應(yīng) DataStream API 的 BoundedOutOfOrdernessTimestampExtractor）

WATERMARK FOR rowtime_column AS rowtime_column - INTERVAL 'n' TIME_UNIT

上文的 SQL 語(yǔ)句中就是設(shè)定了 10 秒的亂序區(qū)間。如果看官對(duì)水印、AscendingTimestampExtractor 和 BoundedOutOfOrdernessTimestampExtractor 不熟的話，可以參見(jiàn)之前的這篇，就能理解為什么會(huì)是這樣的語(yǔ)法了。

https://www.jianshu.com/p/c612e95a5028

下面來(lái)正式建表。

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

執(zhí)行完畢后，我們還可以去到 Hive 執(zhí)行 DESCRIBE FORMATTED ods.streaming_user_active_log 語(yǔ)句，能夠發(fā)現(xiàn)該表并沒(méi)有事實(shí)上的列，而所有屬性（包括 schema、connector、format 等等）都作為元數(shù)據(jù)記錄在了 Hive Metastore 中。

Flink 1.10中SQL、HiveCatalog與事件時(shí)間整合的示例分析

Flink SQL 創(chuàng)建的表都會(huì)帶有一個(gè)標(biāo)記屬性 is_generic=true，圖中未示出。

開(kāi)窗計(jì)算 PV、UV

用30秒的滾動(dòng)窗口，按事件類型來(lái)分組，查詢語(yǔ)句如下。

   
     
   
   
   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 uvFROM rtdw.ods.streaming_user_active_logWHERE platform = 'xyz'GROUP BY eventType, TUMBLE(eventTime, INTERVAL '30' SECOND)

關(guān)于窗口在 SQL 里的表達(dá)方式請(qǐng)參見(jiàn) 官方文檔。1.10 版本 SQL 的官方文檔寫的還是比較可以的。

SQL 文檔：
https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/table/sql/queries.html#group-windows

懶得再輸出到一個(gè)結(jié)果表了，直接轉(zhuǎn)換成流打到屏幕上。

    val queryActiveSql =      """        |......        |......      """.stripMargin    val result = tableEnv.sqlQuery(queryActiveSql)
    result        .toAppendStream[Row]        .print()        .setParallelism(1)

關(guān)于“Flink 1.10中SQL、HiveCatalog與事件時(shí)間整合的示例分析”這篇文章就分享到這里了，希望以上內(nèi)容可以對(duì)大家有一定的幫助，使各位可以學(xué)到更多知識(shí)，如果覺(jué)得文章不錯(cuò)，請(qǐng)把它分享出去讓更多的人看到。

網(wǎng)頁(yè)題目：Flink1.10中SQL、HiveCatalog與事件時(shí)間整合的示例分析
分享URL：http://bm7419.com/article6/igosog.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供品牌網(wǎng)站建設(shè)、建站公司、云服務(wù)器、手機(jī)網(wǎng)站建設(shè)、營(yíng)銷型網(wǎng)站建設(shè)、品牌網(wǎng)站設(shè)計(jì)

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請(qǐng)盡快告知，我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng)，如需處理請(qǐng)聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時(shí)需注明來(lái)源：創(chuàng)新互聯(lián)

猜你還喜歡下面的內(nèi)容

Flink1.10中SQL、HiveCatalog與事件時(shí)間整合的示例分析

Flink1.10中SQL、HiveCatalog與事件時(shí)間整合的示例分析