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

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

• 一是 SQL DDL 對事件時間的支持；
• 二是 Hive Metastore 作為 Flink 的元數據存儲（即 HiveCatalog）。

• 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

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，準備工作就完成了。
      
    
    

   
     
   
   
  


   
     
   
   
  

   
     
   
   
   
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)
   
     
   
   
  


   
     
   
   
  

   
     
   
   
   


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


   
     
   
   
  

   
     
   
   
   
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')
   
     
   
   
  


      
    
    
   
     
   
   
  

• 單調不減水印（對應 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

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

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

 

 

   
     
   
   
   
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)
   
     
   
   
  


      
    
    
   
     
   
   
  

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

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

關于“Flink 1.10中SQL、HiveCatalog與事件時間整合的示例分析”這篇文章就分享到這里了，希望以上內容可以對大家有一定的幫助，使各位可以學到更多知識，如果覺得文章不錯，請把它分享出去讓更多的人看到。