数据仓库 Hive

CREATE TABLE hive_table (
  `id` INT,
  `name` STRING,
  `dt` STRING,
  `hr` STRING
) PARTITIONED BY (dt, hr)
with (
    'connector' = 'hive',  -- Flink 1.13 请使用 'connector' = 'hive'
    'hive-version' = '3.1.1',
    'hive-database' = 'testdb',
    'partition.time-extractor.timestamp-pattern'='$dt $hr:00:00',
    'sink.partition-commit.trigger'='partition-time',
    'sink.partition-commit.delay'='1 h',
    'sink.partition-commit.policy.kind'='metastore,success-file'
);

# 在 Hive 的 testdb 数据库创建 hive_table 数据表
USE testdb;
CREATE TABLE `hive_table` (
  `id` int,
  `name` string)
PARTITIONED BY (`dt` string, `hr` string)
STORED AS ORC;

CREATE TABLE datagen_source_table (
  id INT,
  name STRING,
  log_ts TIMESTAMP(3),
  WATERMARK FOR log_ts AS log_ts - INTERVAL '5' SECOND
) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '10'
);

CREATE TABLE hive_table (
  `id` INT,
  `name` STRING,
  `dt` STRING,
  `hr` STRING
) PARTITIONED BY (dt, hr)
with (
    'connector' = 'hive',  -- Flink 1.13 请使用 'connector' = 'hive'
    'hive-version' = '3.1.1',
    'hive-database' = 'testdb',
    'partition.time-extractor.timestamp-pattern'='$dt $hr:00:00',
    'sink.partition-commit.trigger'='partition-time',
    'sink.partition-commit.delay'='1 h',
    'sink.partition-commit.policy.kind'='metastore,success-file'
);

-- streaming sql, insert into hive table
INSERT INTO hive_table
SELECT id, name, DATE_FORMAT(log_ts, 'yyyy-MM-dd'), DATE_FORMAT(log_ts, 'HH')
FROM datagen_source_table;

-- hive table 作为source 需要显示指定hive中的database，否则会用默认的数据库default_database
CREATE database testdb;

CREATE TABLE `testdb`.hive_table (
  `user_id` INT,
  `first_name` STRING,
  `dt` STRING,
  `hr` STRING  
) PARTITIONED BY (dt, hr) with (
  'connector' = 'hive',
  'hive-version' = '3.1.1',
  'hive-database' = 'testdb',
  'partition.time-extractor.timestamp-pattern' = '$dt $hr:00:00',
  'streaming-source.enable' = 'true',
  'streaming-source.partition.include' = 'all' -- all 、 latest
,
  'streaming-source.monitor-interval' = '1 min',
  'streaming-source.partition-order' = 'partition-name' 
);


CREATE TABLE logger_sink_table (
  `user_id` INT,
  first_name STRING,
  `dt` STRING,
  `hr` STRING  
) WITH (
  'connector' = 'logger',
  'print-identifier' = 'DebugData'
);

INSERT into
  logger_sink_table
SELECT
  *
from
  `testdb`.hive_table;

msck repair table hive_table_xxx;