溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
這篇文章主要介紹了Flink 1.11新特性之SQL Hive Streaming的示例分析，具有一定借鑒價值，感興趣的朋友可以參考下，希望大家閱讀完這篇文章之后大有收獲，下面讓小編帶著大家一起了解一下。
添加相關依賴
<properties><scala.bin.version>2.11</scala.bin.version><flink.version>1.11.0</flink.version><flink-shaded-hadoop.version>2.6.5-10.0</flink-shaded-hadoop.version><hive.version>1.1.0</hive.version></properties><dependencies><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-scala_${scala.bin.version}</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-clients_${scala.bin.version}</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-common</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-connector-hive_${scala.bin.version}</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-sql-connector-kafka_${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.flink</groupId><artifactId>flink-shaded-hadoop-2-uber</artifactId><version>${flink-shaded-hadoop.version}</version></dependency><dependency><groupId>org.apache.hive</groupId><artifactId>hive-exec</artifactId><version>${hive.version}</version></dependency>
創建執行環境
val streamEnv = StreamExecutionEnvironment.getExecutionEnvironmentstreamEnv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)streamEnv.setParallelism(3)val tableEnvSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()val tableEnv = StreamTableEnvironment.create(streamEnv, tableEnvSettings)tableEnv.getConfig.getConfiguration.set(ExecutionCheckpointingOptions.CHECKPOINTING_MODE, CheckpointingMode.EXACTLY_ONCE)tableEnv.getConfig.getConfiguration.set(ExecutionCheckpointingOptions.CHECKPOINTING_INTERVAL, Duration.ofSeconds(20))
注冊 HiveCatalog
val catalogName = "my_catalog"val catalog = new HiveCatalog( catalogName, // catalog name "default", // default database "/Users/lmagic/develop", // Hive config (hive-site.xml) directory "1.1.0" // Hive version)tableEnv.registerCatalog(catalogName, catalog)tableEnv.useCatalog(catalogName)
創建 Kafka 流表
tableEnv.executeSql("CREATE DATABASE IF NOT EXISTS stream_tmp")tableEnv.executeSql("DROP TABLE IF EXISTS stream_tmp.analytics_access_log_kafka")tableEnv.executeSql("""|CREATE TABLE stream_tmp.analytics_access_log_kafka (| ts BIGINT,| userId BIGINT,| eventType STRING,| fromType STRING,| columnType STRING,| siteId BIGINT,| grouponId BIGINT,| partnerId BIGINT,| merchandiseId BIGINT,| procTime AS PROCTIME(),| eventTime AS TO_TIMESTAMP(FROM_UNIXTIME(ts / 1000,'yyyy-MM-dd HH:mm:ss')),| WATERMARK FOR eventTime AS eventTime - INTERVAL '15' SECOND|) WITH (| 'connector' = 'kafka',| 'topic' = 'ods_analytics_access_log',| 'properties.bootstrap.servers' = 'kafka110:9092,kafka111:9092,kafka112:9092'| 'properties.group.id' = 'flink_hive_integration_exp_1',| 'scan.startup.mode' = 'latest-offset',| 'format' = 'json',| 'json.fail-on-missing-field' = 'false',| 'json.ignore-parse-errors' = 'true'|)""".stripMargin)
hive> DESCRIBE FORMATTED stream_tmp.analytics_access_log_kafka;OK# col_name data_type comment# Detailed Table InformationDatabase: stream_tmpOwner: nullCreateTime: Wed Jul 15 18:25:09 CST 2020LastAccessTime: UNKNOWNProtect Mode: NoneRetention: 0Location: hdfs://sht-bdmq-cls/user/hive/warehouse/stream_tmp.db/analytics_access_log_kafkaTable Type: MANAGED_TABLETable Parameters:flink.connector kafkaflink.format jsonflink.json.fail-on-missing-field falseflink.json.ignore-parse-errors trueflink.properties.bootstrap.servers kafka110:9092,kafka111:9092,kafka112:9092flink.properties.group.id flink_hive_integration_exp_1flink.scan.startup.mode latest-offsetflink.schema.0.data-type BIGINTflink.schema.0.name tsflink.schema.1.data-type BIGINTflink.schema.1.name userIdflink.schema.10.data-type TIMESTAMP(3)flink.schema.10.expr TO_TIMESTAMP(FROM_UNIXTIME(`ts` / 1000, 'yyyy-MM-dd HH:mm:ss'))flink.schema.10.name eventTimeflink.schema.2.data-type VARCHAR(2147483647)flink.schema.2.name eventType# 略......flink.schema.9.data-type TIMESTAMP(3) NOT NULLflink.schema.9.expr PROCTIME()flink.schema.9.name procTimeflink.schema.watermark.0.rowtime eventTimeflink.schema.watermark.0.strategy.data-type TIMESTAMP(3)flink.schema.watermark.0.strategy.expr `eventTime` - INTERVAL '15' SECONDflink.topic ods_analytics_access_logis_generic truetransient_lastDdlTime 1594808709# Storage InformationSerDe Library: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDeInputFormat: org.apache.hadoop.mapred.TextInputFormatOutputFormat: org.apache.hadoop.hive.ql.io.IgnoreKeyTextOutputFormatCompressed: NoNum Buckets: -1Bucket Columns: []Sort Columns: []Storage Desc Params:serialization.format 1Time taken: 1.797 seconds, Fetched: 61 row(s)
創建 Hive 表
tableEnv.getConfig.setSqlDialect(SqlDialect.HIVE)tableEnv.executeSql("CREATE DATABASE IF NOT EXISTS hive_tmp")tableEnv.executeSql("DROP TABLE IF EXISTS hive_tmp.analytics_access_log_hive")tableEnv.executeSql("""|CREATE TABLE hive_tmp.analytics_access_log_hive (| ts BIGINT,| user_id BIGINT,| event_type STRING,| from_type STRING,| column_type STRING,| site_id BIGINT,| groupon_id BIGINT,| partner_id BIGINT,| merchandise_id BIGINT|) PARTITIONED BY (| ts_date STRING,| ts_hour STRING,| ts_minute STRING|) STORED AS PARQUET|TBLPROPERTIES (| 'sink.partition-commit.trigger' = 'partition-time',| 'sink.partition-commit.delay' = '1 min',| 'sink.partition-commit.policy.kind' = 'metastore,success-file',| 'partition.time-extractor.timestamp-pattern' = '$ts_date $ts_hour:$ts_minute:00'|)""".stripMargin)
https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/filesystem.html#streaming-sink
流式寫入 Hive
tableEnv.getConfig.setSqlDialect(SqlDialect.DEFAULT)tableEnv.executeSql( """ |INSERT INTO hive_tmp.analytics_access_log_hive |SELECT | ts,userId,eventType,fromType,columnType,siteId,grouponId,partnerId,merchandiseId, | DATE_FORMAT(eventTime,'yyyy-MM-dd'), | DATE_FORMAT(eventTime,'HH'), | DATE_FORMAT(eventTime,'mm') |FROM stream_tmp.analytics_access_log_kafka |WHERE merchandiseId > 0 """.stripMargin)
hive> SELECT from_unixtime(min(cast(ts / 1000 AS BIGINT))),from_unixtime(max(cast(ts / 1000 AS BIGINT))) > FROM hive_tmp.analytics_access_log_hive > WHERE ts_date = '2020-07-15' AND ts_hour = '23' AND ts_minute = '23';OK2020-07-15 23:23:00 2020-07-15 23:23:59Time taken: 1.115 seconds, Fetched: 1 row(s)
流式讀取 Hive
tableEnv.getConfig.getConfiguration.setBoolean(TableConfigOptions.TABLE_DYNAMIC_TABLE_OPTIONS_ENABLED, true)
val result = tableEnv.sqlQuery( """ |SELECT merchandise_id,count(1) AS pv |FROM hive_tmp.analytics_access_log_hive |/*+ OPTIONS( | 'streaming-source.enable' = 'true', | 'streaming-source.monitor-interval' = '1 min', | 'streaming-source.consume-start-offset' = '2020-07-15 23:30:00' |) */ |WHERE event_type = 'shtOpenGoodsDetail' |AND ts_date >= '2020-07-15' |GROUP BY merchandise_id |ORDER BY pv DESC LIMIT 10 """.stripMargin)
result.toRetractStream[Row].print().setParallelism(1)streamEnv.execute()
https://links.jianshu.com/go?to=https%3A%2F%2Fci.apache.org%2Fprojects%2Fflink%2Fflink-docs-release-1.11%2Fdev%2Ftable%2Fhive%2Fhive_streaming.html%23streaming-reading
感謝你能夠認真閱讀完這篇文章，希望小編分享的“Flink 1.11新特性之SQL Hive Streaming的示例分析”這篇文章對大家有幫助，同時也希望大家多多支持億速云，關注億速云行業資訊頻道，更多相關知識等著你來學習!
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
