自定义FileInputFormat始终将一个filesplit分配给一个插槽

Question

我一直在为我们的s3桶编写protobuf记录。我想用flink数据集api从中读取。所以我实现了一个自定义的FileInputFormat来实现这一点。代码如下。

public class ProtobufInputFormat extends FileInputFormat<StandardLog.Pageview> {
    public ProtobufInputFormat() {
    }

    private transient boolean reachedEnd = false;

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

    @Override
    public StandardLog.Pageview nextRecord(StandardLog.Pageview reuse) throws IOException {
        StandardLog.Pageview pageview = StandardLog.Pageview.parseDelimitedFrom(stream);
        if (pageview == null) {
            reachedEnd = true;
        }
        return pageview;
    }

    @Override
    public boolean supportsMultiPaths() {
        return true;
    }
}


public class BatchReadJob {

    public static void main(String... args) throws Exception {

        String readPath1 = args[0];

        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();


        ProtobufInputFormat inputFormat =  new ProtobufInputFormat();
        inputFormat.setNestedFileEnumeration(true);

        inputFormat.setFilePaths(readPath1);

        DataSet<StandardLog.Pageview> dataSource = env.createInput(inputFormat);

        dataSource.map(new MapFunction<StandardLog.Pageview, String>() {
            @Override
            public String map(StandardLog.Pageview value) throws Exception {
                return value.getId();
            }
        }).writeAsText("s3://xxx", FileSystem.WriteMode.OVERWRITE);
        env.execute();

    }

}

问题是flink总是将一个filesplit分配给一个并行槽。换句话说，它总是处理与并行数相同的文件分割数。

我想知道实现自定义FileInputFormat的正确方法是什么。

谢谢。

Answer 1

我相信你看到的行为是因为ExecutionJobVertex用FileInputFormat. createInputSplits()参数调用minNumSplits方法，该参数等于顶点（数据源）并行度。因此，如果您想要一个不同的行为，那么您必须覆盖createInputSplits方法。

虽然你没有说出你真正想要的行为。例如，如果您只想在每个文件中进行一次拆分，那么您可以覆盖testForUnsplittable()子类中的FileInputFormat方法，以便始终返回true;它还应该将（protected）unsplittable布尔值设置为true。