帮我写一个flink读取多个hdfs文件的source

非常好! 下面是一个例子，它展示了如何使用Flink的Hadoop InputFormat API来读取HDFS上的多个文件：

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.hadoop.mapreduce.HadoopInputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;

// 获取执行环境
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

// 创建Hadoop Job
Job job = Job.getInstance();

// 设置输入格式为TextInputFormat，这意味着我们将读取文本文件
FileInputFormat.setInputPaths(job, "/path/to/input/*");
job.setInputFormatClass(TextInputFormat.class);

// 创建Flink DataSet，使用Hadoop InputFormat
DataSet<String> inputData = env.createInput(new HadoopInputFormat<>(job, TextInputFormat.class));

// 对读取的每一行执行map操作
DataSet<String> mappedData = inputData.map(new MapFunction<String, String>() {
  public String map(String value) {
    // 对每一行执行你想要的操作
    return value;
  }
});

// 使用DataSet API进行其他操作，例如groupBy，filter等等

在这个例子中，我们使用了Hadoop的TextInputFormat来读取HDFS上的文本文件。你也可以使用其他的Hadoop InputFormat，例如SequenceFileInputFormat或AvroInputFormat等等。

希望这个例子能帮到你！