flink消费kafka数据后 再写入hive

可以通过Flink的Hive Sink将消费的Kafka数据写入Hive中。具体步骤如下：

1. 在Flink中创建Kafka数据源，并使用Flink的DataStream API将数据转换为需要写入Hive的格式。

2. 使用Flink的Hive Sink将数据写入Hive中。Hive Sink需要配置Hive JDBC连接信息和表结构信息。

下面是一个示例代码，可以根据实际情况进行修改：

// 创建Flink的执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 创建Kafka数据源
Properties props = new Properties();
props.setProperty("bootstrap.servers", "localhost:9092");
props.setProperty("group.id", "test-group");
FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("test-topic", new SimpleStringSchema(), props);

// 读取Kafka数据
DataStream<String> stream = env.addSource(consumer);

// 转换数据格式
DataStream<Tuple2<String, Integer>> data = stream.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
    @Override
    public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
        // 根据实际情况进行数据转换
        String[] parts = s.split(",");
        collector.collect(new Tuple2<>(parts[0], Integer.parseInt(parts[1])));
    }
});

// 配置Hive Sink
HiveConf hiveConf = new HiveConf();
hiveConf.addResource(new Path("/path/to/hive-site.xml"));
HiveCatalog hiveCatalog = new HiveCatalog("myHiveCatalog", "default", hiveConf);
env.registerCatalog("myHiveCatalog", hiveCatalog);
env.useCatalog("myHiveCatalog");

HiveTableSink hiveTableSink = new HiveTableSink.Builder()
    .setPartitions(Collections.singletonList("dt='20220101'"))
    .setTable(new ObjectPath("default", "test"))
    .setHiveConf(hiveConf)
    .setOverwrite(true)
    .setStaticPartitionInsert(false)
    .build();

// 写入Hive
data.addSink(hiveTableSink);

// 执行任务
env.execute("Kafka to Hive");

需要注意的是，在执行程序前需要先创建好Hive表，并且表的结构要与代码中配置的一致。另外，Hive Sink在写入数据时会自动创建分区，因此在写入数据前需要确保分区已经存在。