关于FlinkSQL从kafka读取数据写到hive的一些问题

Mon, 01 Nov 2021 00:05:53 -0700 

如题，我看了官方文档，定义好kafka和hive表。
写的时候提示要指定提交策略，就又看了看文档，如下为文档实例。

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 -- Define
watermark on TIMESTAMP column) WITH (...);


如上，如果是这样的话，那就会出现个问题。所有需要写入的hive表其实都需要重新定义一次，部分原先的表是hive中定义的。现在我需要重新定义一次可能。

其次，为了避免重新定义表有问题啥的，我可能会重新定义另一个数据库中同名表，但指定到和hive表相同的存储路径。
但如果hive中修改原表，我这边不改变flink hive表定义，又会出现不一致的情况。


此外，flink这样定义的hive表和hive自己定义的肯定意义一致吗，不会影响hive自身的读写吧。

回复