Writing

Flink支持在批处理和流模式下从Hive写入数据。当作为批处理应用程序运行时，Flink将只在作业完成时才将这些记录写入Hive表。批写既支持追加现有表，也支持重写现有表。

# ------ INSERT INTO将追加到表或分区上，保持现有数据不变------ 
Flink SQL> INSERT INTO mytable SELECT 'Tom', 25;

# ------ INSERT OVERWRITE将覆盖表或分区中的任何现有数据------ 
Flink SQL> INSERT OVERWRITE mytable SELECT 'Tom', 25;

数据也可以插入到特定的分区

# ------ 插入静态分区------ 
Flink SQL> INSERT OVERWRITE myparttable PARTITION (my_type='type_1', my_date='2019-08-08') SELECT 'Tom', 25;

# ------ 使用动态分区插入 ------ 
Flink SQL> INSERT OVERWRITE myparttable SELECT 'Tom', 25, 'type_1', '2019-08-08';

# ------使用静态(my_type)和动态(my_date)分区插入 ------ 
Flink SQL> INSERT OVERWRITE myparttable PARTITION (my_type='type_1') SELECT 'Tom', 25, '2019-08-08';

流写不断地向Hive添加新数据，以增量的方式提交记录——使其可见。用户通过几个属性控制何时/如何触发提交。流写不支持插入覆盖。
下面展示了如何使用流接收器写一个流查询，将数据从Kafka写入到Hive表中，并提交分区，然后运行一个批量查询将数据读出来。
有关可用配置的完整列表，请参阅流接收器。
https://ci.apache.org/projects/flink/flink-docs-release-1.12/zh/dev/table/connectors/filesystem.html#streaming-sink

SET table.sql-dialect=hive;
CREATE TABLE hive_table (
  user_id STRING,
  order_amount DOUBLE
) PARTITIONED BY (dt STRING, hr STRING) STORED AS parquet TBLPROPERTIES (
  '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'
);

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
) WITH (...);

-- streaming sql, insert into hive table
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;

-- batch sql, select with partition pruning
SELECT * FROM hive_table WHERE dt='2020-05-20' and hr='12';