Flink 读取 本地日志文件

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class ReadLocalLogs {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度为1（便于调试）
        env.setParallelism(1);
        String logPath = "/path/to/local/logs";
        DataStream<String> logs = env.readTextFile(logPath);
        // 对每条日志进行处理操作
        logs.print();
        env.execute("Read Local Logs");
    }
}

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import java.io.*;
public class CustomDataSource implements RichSourceFunction<String> {
    private volatile boolean isRunning = true;
    private BufferedReader reader;
    @Override
    public void open(Configuration parameters) throws Exception {
        File file = new File("/path/to/local/logs");
        InputStream inputStream = new FileInputStream(file);
        reader = new BufferedReader(new InputStreamReader(inputStream));
    }
    @Override
    public void run(SourceContext<String> ctx) throws Exception {
        while (isRunning && !Thread.currentThread().isInterrupted()) {
            String line = reader.readLine();
            if (line != null) {
                ctx.collect(line);
            } else {
                Thread.sleep(500);
            }
        }
    }
    @Override
    public void cancel() {
        isRunning = false;
    }
}
public class MainClass {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); 
        // 创建自定义数据源
        DataStream<String> logs = env.addSource(new CustomDataSource()); 
        // 对每条日志进行处理操作
        logs.print();
        env.execute("Custom Source Example");
    }
}