flink 实时同步kafka数据到hdfs

Apache Flink 是一个流处理框架，支持实时数据处理和批处理。Flink 可以轻松地与 Apache Kafka 集成，实现从 Kafka 中读取数据并将其写入 HDFS。

下面是实现实时同步 Kafka 数据到 HDFS 的基本步骤：

1. 在 Flink 中引入 Kafka 和 HDFS 的依赖。

2. 创建一个 Flink StreamExecutionEnvironment 对象，并设置相关参数。

3. 创建一个 Kafka 数据源，并从 Kafka 中读取数据。

4. 对读取的数据进行转换和处理。

5. 将处理后的数据写入 HDFS 中。

以下是一个基本的示例代码：

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.core.fs.FileSystem.WriteMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer09;

public class KafkaToHDFS {

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

        // 从命令行参数中读取参数
        final ParameterTool params = ParameterTool.fromArgs(args);

        // 创建一个 Flink StreamExecutionEnvironment 对象，并设置相关参数
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(params.getInt("parallelism", 1));

        // 设置 Kafka 数据源
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "localhost:9092");
        props.setProperty("group.id", "test");
        FlinkKafkaConsumer09<String> consumer = new FlinkKafkaConsumer09<>(
                params.getRequired("topic"), new SimpleStringSchema(), props);

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

        // 对读取的数据进行转换和处理
        DataStream<String> transformed = stream.map(new MapFunction<String, String>() {
            @Override
            public String map(String value) throws Exception {
                // 进行转换和处理
                return value.toUpperCase();
            }
        });

        // 将处理后的数据写入 HDFS 中
        transformed.writeAsText(params.getRequired("output"), WriteMode.OVERWRITE);

        // 执行任务
        env.execute("KafkaToHDFS");
    }
}

在执行上述代码之前，需要先将 Flink 的依赖添加到项目中，并修改示例代码中的相关配置参数，如 Kafka 的连接地址、topic 名称和 HDFS 的输出路径等。