flink读取目录写往hdfs，写个样例出来

好的，以下是一个简单的 Flink 代码示例，用于读取目录并将其写入 HDFS：

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.fs.StringWriter;
import org.apache.flink.streaming.connectors.fs.bucketing.BucketingSink;
import org.apache.flink.util.Collector;

public class FlinkHdfsExample {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 读取目录
        DataStream<String> input = env.readTextFile("/path/to/directory");

        // 将数据写入 HDFS
        BucketingSink<String> sink = new BucketingSink<>("hdfs://localhost:9000/path/to/output");
        sink.setWriter(new StringWriter<>());
        sink.setBatchSize(1024 * 1024 * 128); // 128 MB
        sink.setBatchRolloverInterval(60 * 1000); // 1 minute
        sink.setInactiveBucketCheckInterval(60 * 1000); // 1 minute

        input.addSink(sink);

        env.execute("FlinkHdfsExample");
    }
}

希望这个示例能够帮助你解决问题！