flink1.16 数据写入hive样例代码

以下是使用 Apache Flink 1.16 将数据写入 Hive 的样例代码：

// 导入相关依赖
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.catalog.hive.HiveCatalog;
import org.apache.flink.types.Row;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

// 创建 Flink 流处理环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);

// 创建 Flink 表环境
EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

// 创建 HiveCatalog
String catalogName = "my_catalog";
String defaultDatabase = "my_database";
String hiveConfDir = "path/to/hive/conf/dir";
HiveCatalog hiveCatalog = new HiveCatalog(catalogName, defaultDatabase, hiveConfDir);
tableEnv.registerCatalog(catalogName, hiveCatalog);
tableEnv.useCatalog(catalogName);

// 创建表并注册到 HiveCatalog 中
String createTable = "CREATE TABLE my_table (id INT, name STRING) PARTITIONED BY (dt STRING) STORED AS PARQUET";
tableEnv.executeSql(createTable);

// 将数据写入 Hive 表
tableEnv.toRetractStream(tableEnv.sqlQuery("SELECT * FROM my_source"), Row.class)
    .addSink(new SinkFunction<Tuple2<Boolean, Row>>() {
        @Override
        public void invoke(Tuple2<Boolean, Row> value) throws Exception {
            if (value.f0) {
                tableEnv.executeSql("INSERT INTO my_table SELECT *, DATE_FORMAT(NOW(), 'yyyy-MM-dd') AS dt FROM my_source");
            }
        }
    });

// 执行任务
env.execute();

其中，`my_catalog`、`my_database`、`my_table`、`my_source` 都需要替换成你自己的表名称和数据源。此外，`hiveConfDir` 需要替换成你的 Hive 配置文件所在的路径。