生成flink消费kafka数据，以orc格式写入到hdfs

以下是一个基于Flink消费Kafka数据并将其写入HDFS的示例：

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.formats.orc.OrcSplitReaderUtil;
import org.apache.flink.formats.orc.OrcWriterFactory;
import org.apache.flink.formats.orc.vector.StringColumnVector;
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.JSONKeyValueDeserializationSchema;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.*;
import org.apache.flink.types.Row;

import java.util.Properties;

public class FlinkKafkaHdfsOrcDemo {

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

        // set up the streaming execution environment
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1); // set parallelism to 1 for demo purposes

        // set up the Kafka consumer properties
        Properties kafkaProps = new Properties();
        kafkaProps.setProperty("bootstrap.servers", "localhost:9092");
        kafkaProps.setProperty("group.id", "flink-kafka-consumer-group");

        // create a FlinkKafkaConsumer instance to consume Kafka data
        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>("my-topic", new SimpleStringSchema(), kafkaProps);

        // create a data stream from the Kafka source
        DataStream<String> kafkaStream = env.addSource(kafkaConsumer);

        // parse the JSON data and create a table from it
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().useBlinkPlanner().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);
        tableEnv.connect(new Kafka().version("universal").topic("my-topic").startFromEarliest().property("bootstrap.servers", "localhost:9092").property("group.id", "flink-kafka-consumer-group"))
                .withFormat(new Json().deriveSchema())
                .withSchema(new Schema().field("name", DataTypes.STRING()).field("age", DataTypes.INT()))
                .createTemporaryTable("myTable");
        Table myTable = tableEnv.from("myTable");

        // create an OrcWriterFactory to write ORC data
        OrcWriterFactory<Row> orcWriterFactory = (OrcWriterFactory<Row>) OrcSplitReaderUtil.createRowOrcWriterFactory(
                new String[]{"name", "age"},
                new OrcSplitReaderUtil.TypeDescription[]{
                        OrcSplitReaderUtil.TypeDescription.createString(),
                        OrcSplitReaderUtil.TypeDescription.createInt()
                });

        // create a FlinkKafkaProducer instance to write Kafka data
        FlinkKafkaProducer<Row> kafkaProducer = new FlinkKafkaProducer<>(
                "my-topic",
                new OrcRowSerializationSchema("/path/to/hdfs/file.orc", orcWriterFactory),
                kafkaProps,
                FlinkKafkaProducer.Semantic.EXACTLY_ONCE);

        // write the table data to HDFS in ORC format
        myTable.execute().output(kafkaProducer);

        // execute the job
        env.execute("Flink Kafka HDFS ORC Demo");
    }

    // implementation of OrcRowSerializationSchema
    private static class OrcRowSerializationSchema implements FlinkKafkaProducer.SerializationSchema<Row> {

        private final String filePath;
        private final OrcWriterFactory<Row> orcWriterFactory;
        private transient OrcWriterFactory.Writer<Row> orcWriter;

        public OrcRowSerializationSchema(String filePath, OrcWriterFactory<Row> orcWriterFactory) {
            this.filePath = filePath;
            this.orcWriterFactory = orcWriterFactory;
        }

        @Override
        public byte[] serialize(Row row) {
            try {
                if (orcWriter == null) {
                    orcWriter = orcWriterFactory.createWriter(filePath, FileSystem.getHadoopFileSystem(new org.apache.flink.core.fs.Path(filePath).toUri()), true);
                }
                StringColumnVector nameVector = new StringColumnVector(1);
                nameVector.vector[0] = row.getField(0).toString();
                StringColumnVector ageVector = new StringColumnVector(1);
                ageVector.vector[0] = row.getField(1).toString();
                orcWriter.addRow(nameVector, ageVector);
                return null;
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
    }
}

该示例使用Flink的Table API从Kafka消费数据，并将其写入HDFS中的ORC文件。示例代码使用`JsonKeyValueDeserializationSchema`解析JSON格式的数据，并使用`OrcWriterFactory`将数据写入ORC文件。在示例中，`OrcWriterFactory`被配置为使用String和Int类型的列。还创建了一个`OrcRowSerializationSchema`类，它将Flink的`Row`类型转换为ORC文件中的列向量，并使用`OrcWriterFactory.Writer`将数据写入ORC文件。

注意：在实际使用中，应该根据实际需求修改示例代码，并根据需要添加适当的错误处理和容错机制。