org.apache.flink.connectors.hive.FlinkHiveException: Streaming write to partitioned hive table `myhive_default`.`default`.`t_amm_23g` without providing a commit policy. Make sure to set a proper value for sink.partition-commit.policy.kind

这个异常是因为在执行 Flink 写入 Hive 分区表时，没有提供正确的提交策略。可以通过设置 `sink.partition-commit.policy.kind` 属性来解决这个问题。

可以尝试将 `sink.partition-commit.policy.kind` 设置为以下值之一：

- `metastore`: 在元数据存储中提交分区
- `success-file`: 在成功文件中提交分区
- `filesystem`: 在文件系统中提交分区

例如，将 `sink.partition-commit.policy.kind` 设置为 `metastore`：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);

DataStream<Tuple2<String, Integer>> stream = env.fromElements(
    Tuple2.of("foo", 1),
    Tuple2.of("bar", 2)
);

HiveCatalog hiveCatalog = new HiveCatalog("myhive", "default", "/path/to/hive/conf");
hiveCatalog.open();

TableSchema schema = new TableSchema(
    new String[] {"name", "count"},
    new TypeInformation<?>[] {Types.STRING, Types.INT}
);

HiveTableSink sink = new HiveTableSink(
    hiveCatalog,
    new CatalogTable(
        schema,
        new HashMap<String, String>(),
        "",
        "",
        TableType.EXTERNAL
    ),
    new String[]{"name"},
    new HiveOptions.Builder().setPartitionCommitPolicyKind("metastore").build()
);

stream.writeToSink(sink);

env.execute();

注意：具体的提交策略设置需要根据具体情况进行调整。