flink如何监听目录下的文件

Flink可以使用`FileSource`来监听目录下的文件。具体操作如下：

1. 导入必要的依赖：

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.FileProcessingMode;
import org.apache.flink.streaming.api.functions.source.FileSource;
import org.apache.flink.streaming.api.functions.source.TimestampedFileInputSplit;
import org.apache.flink.util.Collector;

2. 创建`StreamExecutionEnvironment`：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

3. 创建`FileSource`：

String filePath = "/path/to/directory";
FileSource<String> fileSource = new FileSource<>(filePath, new TextInputFormat(), FileProcessingMode.PROCESS_CONTINUOUSLY, 1000);

其中，`filePath`是目标目录的路径，`TextInputFormat`是指读取文本文件，`FileProcessingMode.PROCESS_CONTINUOUSLY`表示持续监听目录，`1000`是指每隔1秒扫描一次目录。

4. 对文件内容进行处理：

DataStream<String> lines = env.readFile(fileSource, filePath, FileProcessingMode.PROCESS_CONTINUOUSLY, 1000)
                .flatMap(new FlatMapFunction<String, String>() {
                    @Override
                    public void flatMap(String s, Collector<String> collector) throws Exception {
                        String[] words = s.split("\\W+");
                        for (String word : words) {
                            if (word.length() > 0) {
                                collector.collect(word);
                            }
                        }
                    }
                });

其中，`env.readFile()`方法会返回`DataStream`，我们可以在其中使用`flatMap`等操作对文件内容进行处理。

5. 打印结果：

lines.print();

6. 启动任务：

env.execute("File Streaming Word Count");

完整代码示例：

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.FileProcessingMode;
import org.apache.flink.streaming.api.functions.source.FileSource;
import org.apache.flink.streaming.api.functions.source.TimestampedFileInputSplit;
import org.apache.flink.streaming.api.functions.source.FileProcessingMode.WatchType;
import org.apache.flink.streaming.api.functions.source.FileProcessingMode.WatchType;
import org.apache.flink.streaming.api.functions.source.FileProcessingMode.WatchType;
import org.apache.flink.streaming.api.functions.source.FileProcessingMode.WatchType;
import org.apache.flink.streaming.api.functions.source.FileProcessingMode.WatchType;
import org.apache.flink.streaming.api.functions.source.FileProcessingMode.WatchType;
import org.apache.flink.streaming.api.functions.source.FileProcessingMode.WatchType;
import org.apache.flink.streaming.api.functions.source.FileProcessingMode.WatchType;
import org.apache.flink.streaming.api.functions.source.FileProcessingMode.WatchType;
import org.apache.flink.streaming.api.functions.source.FileProcessingMode.WatchType;
import org.apache.flink.streaming.api.functions.source.FileProcessingMode.WatchType;
import org.apache.flink.streaming.api.functions.source.FileProcessingMode.WatchType;
import org.apache.flink.util.Collector;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;


import java.util.Properties;

public class FileStreamingWordCount {

    public static void main(String[] args) throws Exception {
        // 创建 execution environment
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置并行度为1，方便观察结果
        env.setParallelism(1);

        // 监听文件目录
        String filePath = "/path/to/directory";
        FileSource<String> fileSource = new FileSource<>(filePath, new TextInputFormat(), FileProcessingMode.PROCESS_CONTINUOUSLY, 1000);

        // 处理文件内容
        DataStream<String> lines = env.readFile(fileSource, filePath, FileProcessingMode.PROCESS_CONTINUOUSLY, 1000)
                .flatMap(new FlatMapFunction<String, String>() {
                    @Override
                    public void flatMap(String s, Collector<String> collector) throws Exception {
                        String[] words = s.split("\\W+");
                        for (String word : words) {
                            if (word.length() > 0) {
                                collector.collect(word);
                            }
                        }
                    }
                });

        // 统计单词出现次数
        DataStream<Tuple2<String, Integer>> wordCounts = lines
                .map(word -> Tuple2.of(word, 1))
                .keyBy(0)
                .sum(1);

        // 打印结果
        wordCounts.print();

        // 将结果写入 Kafka
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");
        FlinkKafkaProducer<String> producer = new FlinkKafkaProducer<>("word_count", new SimpleStringSchema(), properties);
        producer.setWriteTimestampToKafka(true);
        wordCounts.map(tuple -> tuple.toString()).addSink(producer);

        // 执行任务
        env.execute("File Streaming Word Count");
    }
}