java在flink流处理中一次读取两行文件

Question

我想用一个flink流处理文件，其中两行属于同一行。第一行是标题，第二行是相应的文本

这些文件位于我的本地文件系统中。我将readFile(fileInputFormat, path, watchType, interval, pathFilter, typeInfo)方法与自定义FileInputFormat一起使用

我的流媒体作业类如下所示：

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<Read> inputStream = env.readFile(new ReadInputFormatTest("path/to/monitored/folder"), "path/to/monitored/folder", FileProcessingMode.PROCESS_CONTINUOUSLY, 100);
inputStream.print();
env.execute("Flink Streaming Java API Skeleton");

我的{}是这样的：

public class ReadInputFormatTest extends FileInputFormat<Read> {

    private transient FileSystem fileSystem;
    private transient BufferedReader reader;
    private final String inputPath;
    private String headerLine;
    private String readLine;

    public ReadInputFormatTest(String inputPath) {
        this.inputPath = inputPath;
    }

    @Override
    public void open(FileInputSplit inputSplit) throws IOException {
        FileSystem fileSystem = getFileSystem();
        this.reader = new BufferedReader(new InputStreamReader(fileSystem.open(inputSplit.getPath())));
        this.headerLine = reader.readLine();
        this.readLine = reader.readLine();
    }

    private FileSystem getFileSystem() {
        if (fileSystem == null) {
            try {
                fileSystem = FileSystem.get(new URI(inputPath));
            } catch (URISyntaxException | IOException e) {
                throw new RuntimeException(e);
            }
        }
        return fileSystem;
    }

    @Override
    public boolean reachedEnd() throws IOException {
        return headerLine == null;
    }

    @Override
    public Read nextRecord(Read r) throws IOException {
        r.setHeader(headerLine);
        r.setSequence(readLine);

        headerLine = reader.readLine();
        readLine = reader.readLine();

        return r;
    }
}

正如预期的那样，标题和文本一起存储在一个对象中。但是，该文件被读取八次。所以问题是并行化。在何处以及如何指定一个文件只处理一次，而多个文件并行处理？ 或者我必须进一步更改我的自定义FileInputFormat

Answer 1

我会修改您的源代码以发出可用的文件名（而不是实际的文件内容），然后添加一个新处理器从输入流中读取名称，然后发出成对的行。换言之，将电流源拆分为一个源，然后是一个处理器。处理器可以以任何并行度运行，源代码将是单个实例