傳統(tǒng)的離線數(shù)倉是由 Hive 加上 HDFS 的方案，Hive 數(shù)倉有著成熟和穩(wěn)定的大數(shù)據(jù)分析能力，結(jié)合調(diào)度和上下游工具，構(gòu)建一個完整的數(shù)據(jù)處理分析平臺，流程如下：

• Flume 把數(shù)據(jù)導入 Hive 數(shù)倉

• 調(diào)度工具，調(diào)度 ETL 作業(yè)進行數(shù)據(jù)處理

• 在 Hive 數(shù)倉的表上，可以進行靈活的 Ad-hoc 查詢

• 調(diào)度工具，調(diào)度聚合作業(yè)輸出到BI層的數(shù)據(jù)庫中

這個流程下的問題是：

• 導入過程不夠靈活，這應該是一個靈活 SQL 流計算的過程

• 基于調(diào)度作業(yè)的級聯(lián)計算，實時性太差

• ETL 不能有流式的增量計算

針對離線數(shù)倉的特點，隨著實時計算的流行，越來越多的公司引入實時數(shù)倉，實時數(shù)倉基于 Kafka + Flink streaming，定義全流程的流計算作業(yè)，有著秒級甚至毫秒的實時性。

但是，實時數(shù)倉的一個問題是歷史數(shù)據(jù)只有 3-15 天，無法在其上做 Ad-hoc 的查詢。如果搭建 Lambda 的離線+實時的架構(gòu)，維護成本、計算存儲成本、一致性保證、重復的開發(fā)會帶來很大的負擔。

-- 結(jié)合Hive dialect使用Hive DDL語法SET table.sql-dialect=hive;CREATE TABLE hive_table (  user_id STRING,  order_amount DOUBLE) PARTITIONED BY (  dt STRING,  hour STRING) STORED AS PARQUET TBLPROPERTIES (  -- 使用partition中抽取時間，加上watermark決定partiton commit的時機  'sink.partition-commit.trigger'='partition-time',  -- 配置hour級別的partition時間抽取策略，這個例子中dt字段是yyyy-MM-dd格式的天，hour是0-23的小時，timestamp-pattern定義了如何從這兩個partition字段推出完整的timestamp  'partition.time-extractor.timestamp-pattern'=’$dt $hour:00:00’,  -- 配置dalay為小時級，當 watermark > partition時間 + 1小時，會commit這個partition  'sink.partition-commit.delay'='1 h',  -- partitiion commit的策略是：先更新metastore(addPartition)，再寫SUCCESS文件  'sink.partition-commit.policy.kind’='metastore,success-file') SET table.sql-dialect=default;CREATE TABLE kafka_table (  user_id STRING,  order_amount DOUBLE,  log_ts TIMESTAMP(3),  WATERMARK FOR log_ts AS log_ts - INTERVAL '5' SECOND) -- 可以結(jié)合Table Hints動態(tài)指定table properties [3]INSERT INTO TABLE hive_table SELECT user_id, order_amount, DATE_FORMAT(log_ts, 'yyyy-MM-dd'), DATE_FORMAT(log_ts, 'HH') FROM kafka_table;

SELECT * FROM hive_table/*+ OPTIONS('streaming-source.enable'=’true’,'streaming-source.consume-start-offset'='2020-05-20') */;

SELECT  o.amout, o.currency, r.rate, o.amount * r.rateFROM  Orders AS o  JOIN LatestRates FOR SYSTEM_TIME AS OF o.proctime AS r  ON r.currency = o.currency

CREATE TABLE fs_table (  user_id STRING,  order_amount DOUBLE,  dt STRING,  hour STRING) PARTITIONED BY (dt, hour) WITH (  ’connector’=’filesystem’,  ’path’=’...’,  ’format’=’parquet’,  'partition.time-extractor.timestamp-pattern'=’$dt $hour:00:00’,  'sink.partition-commit.delay'='1 h',  ‘sink.partition-commit.policy.kind’='success-file'))
-- stream environment or batch environmentINSERT INTO TABLE fs_table SELECT user_id, order_amount, DATE_FORMAT(log_ts, 'yyyy-MM-dd'), DATE_FORMAT(log_ts, 'HH') FROM kafka_table;
-- 通過 Partition 查詢SELECT * FROM fs_table WHERE dt=’2020-05-20’ and hour=’12’;

slotmanager.number-of-slots.max