使用java编写flink消费kafka写入hive的demo

以下是使用Java编写Flink消费Kafka写入Hive的示例代码：

1. 导入依赖

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchemaWrapper;
import org.apache.flink.streaming.util.serialization.SimpleStringSchema;

2. 配置Kafka连接

String kafkaBootstrapServers = "localhost:9092";
String kafkaTopic = "test";
Properties kafkaProps = new Properties();
kafkaProps.setProperty("bootstrap.servers", kafkaBootstrapServers);
kafkaProps.setProperty("group.id", "flink-group");

3. 创建 Flink 环境和 Kafka 消费者

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<String> kafkaStream = env.addSource(new FlinkKafkaConsumer<>(kafkaTopic, new SimpleStringSchema(), kafkaProps));

4. 对收到的消息进行处理

DataStream<String> processedStream = kafkaStream.map(new MapFunction<String, String>() {
    @Override
    public String map(String value) throws Exception {
        // 在这里对数据进行处理，返回处理后的数据
        return value;
    }
});

5. 将处理后的数据写入 Hive

String hiveTableName = "test";
String hiveMetastoreUri = "thrift://localhost:9083";
String hiveDbName = "default";
String hivePartitionColumn = "dt";
String hivePartitionValue = "20220101";
String hiveOutputPath = "/user/hive/warehouse/" + hiveDbName + ".db/" + hiveTableName + "/" + hivePartitionColumn + "=" + hivePartitionValue;

DataStream<String> hiveDataStream = processedStream.map(new MapFunction<String, String>() {
    @Override
    public String map(String value) throws Exception {
        // 在这里将数据转换为 Hive 表的格式，返回转换后的数据
        return value;
    }
});

// 将数据写入 Hive
hiveDataStream.addSink(new FlinkHiveOutputFormat<>(new Path(hiveOutputPath), new org.apache.hadoop.hive.ql.io.orc.OrcSerde(), new Object[]{}));

6. 将处理后的数据写回 Kafka

String kafkaOutputTopic = "output";
FlinkKafkaProducer<String> kafkaProducer = new FlinkKafkaProducer<>(kafkaBootstrapServers, kafkaOutputTopic, new KeyedSerializationSchemaWrapper<>(new SimpleStringSchema()), kafkaProps);
// 将数据写回 Kafka
processedStream.addSink(kafkaProducer);

完整示例代码：

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchemaWrapper;
import org.apache.flink.streaming.util.serialization.SimpleStringSchema;

import java.util.Properties;

public class FlinkKafkaToHiveDemo {

    public static void main(String[] args) throws Exception {

        String kafkaBootstrapServers = "localhost:9092";
        String kafkaTopic = "test";
        Properties kafkaProps = new Properties();
        kafkaProps.setProperty("bootstrap.servers", kafkaBootstrapServers);
        kafkaProps.setProperty("group.id", "flink-group");

        String hiveTableName = "test";
        String hiveMetastoreUri = "thrift://localhost:9083";
        String hiveDbName = "default";
        String hivePartitionColumn = "dt";
        String hivePartitionValue = "20220101";
        String hiveOutputPath = "/user/hive/warehouse/" + hiveDbName + ".db/" + hiveTableName + "/" + hivePartitionColumn + "=" + hivePartitionValue;

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<String> kafkaStream = env.addSource(new FlinkKafkaConsumer<>(kafkaTopic, new SimpleStringSchema(), kafkaProps));

        DataStream<String> processedStream = kafkaStream.map(new MapFunction<String, String>() {
            @Override
            public String map(String value) throws Exception {
                // 在这里对数据进行处理，返回处理后的数据
                return value;
            }
        });

        DataStream<String> hiveDataStream = processedStream.map(new MapFunction<String, String>() {
            @Override
            public String map(String value) throws Exception {
                // 在这里将数据转换为 Hive 表的格式，返回转换后的数据
                return value;
            }
        });

        DataStream<String> kafkaOutputStream = processedStream.map(new MapFunction<String, String>() {
            @Override
            public String map(String value) throws Exception {
                // 在这里对数据进行处理，返回处理后的数据
                return value;
            }
        });

        FlinkKafkaProducer<String> kafkaProducer = new FlinkKafkaProducer<>(kafkaBootstrapServers, kafkaOutputTopic, new KeyedSerializationSchemaWrapper<>(new SimpleStringSchema()), kafkaProps);
        processedStream.addSink(kafkaProducer);

        hiveDataStream.addSink(new FlinkHiveOutputFormat<>(new Path(hiveOutputPath), new org.apache.hadoop.hive.ql.io.orc.OrcSerde(), new Object[]{}));

        env.execute("FlinkKafkaToHiveDemo");
    }
}