Apache Flink 是一个流处理框架，而 Hadoop 是一个分布式存储和计算框架。要在 Flink 中使用 Hadoop 进行数据清洗，你需要将 Flink 与 Hadoop 集成。以下是一个简单的步骤来执行此操作：
<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-hadoop_2.11</artifactId> <version>${flink.version}</version> </dependency> 请将 ${flink.version} 替换为你的 Flink 版本，例如 1.12.0。
import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.connectors.hadoop.HadoopInputFormat; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; public class DataCleaningJob { public static void main(String[] args) throws Exception { final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 设置 Hadoop 配置 Configuration hadoopConf = new Configuration(); hadoopConf.set("fs.defaultFS", "hdfs://localhost:9000"); // 读取 Hadoop 上的数据 DataStream<String> input = env.readFile( new HadoopInputFormat<>(new Path("hdfs://localhost:9000/input"), TextInputFormat.class, hadoopConf), "/input", FileProcessingMode.PROCESS_CONTINUOUSLY, 1000 ); // 数据清洗：删除空行和转换为大写 DataStream<String> cleanedData = input .filter(new MapFunction<String, String>() { @Override public String map(String value) throws Exception { return value != null && !value.trim().isEmpty() ? value.toUpperCase() : null; } }) .filter(value -> value != null); // 将清洗后的数据写入 Hadoop cleanedData.addSink(new HadoopOutputFormat<>(new Path("hdfs://localhost:9000/output"), TextOutputFormat.class, hadoopConf)); env.execute("Data Cleaning Job"); } } 在这个示例中，我们首先创建了一个 Flink 作业，然后设置了 Hadoop 配置。接下来，我们使用 HadoopInputFormat 从 Hadoop 读取数据。然后，我们使用 filter 函数删除空行并将所有文本转换为大写。最后，我们使用 HadoopOutputFormat 将清洗后的数据写入 Hadoop。
请注意，这个示例仅用于演示目的。实际的数据清洗操作可能会根据你的需求和数据源而有所不同。你可以根据需要修改 Flink 作业以满足你的数据清洗需求。
