apache-flink - Apache Flink - 为输入分配唯一 ID

Question

我正在加载 CSV 文件并使用自定义映射函数将每一行转换为 POJO。对于我的程序逻辑，我需要为每个 POJO 提供一个从 0 到 n 的唯一 ID(其中 n 是总行号)。我的问题是，我可以使用转换函数为每个 POJO 分配一个唯一的 ID(例如初始行号)吗？理想的方法是在 UDF 中获取 Iterable 并在遍历输入元组时递增变量，最后输出相应的 POJO。我的代码目前看起来像这样:

DataSet<MyType> input = env.readCsvFile("/path/file.csv")
  .includeFields("1111")
  .types(String.class, Double.class, Double.class,Double.class)
  .map(new ParseData());

其中 ParseData 将元组转换为 MyType POJO。

是否有实现此任务的最佳实践？

Answer 1

棘手的部分是，您在分布式系统中运行代码，因此 ParseData 函数的并行实例彼此独立运行。

您仍然可以使用 ParseData 中的本地 ID 计数器分配唯一 ID。避免重复的技巧是正确的初始化和计数器递增。假设您有四个并行度，您将获得四个 ParseData 实例(我们称它们为 PD1 ... PD4)。您将执行以下 ID 分配:

PD1: 0, 4, 8, 12, ...
PD2: 1, 5, 9, 13, ...
PD3, 2, 6, 10, 14, ...
PD4: 3, 7, 11, 15, ...

您可以通过使用不同的值(详情如下)初始化并行实例并根据您的并行度(即 ID += parallelism)增加每个实例的计数来实现此目的。

在 Flink 中，并行函数的所有实例都会自动分配一个唯一的编号(所谓的任务索引)。您可以只使用此号码来初始化您的 ID 计数器。您可以通过 RuntimeContext.getIndexOfThisSubtask() 获取任务索引。您还可以通过 RuntimeContext.getNumberOfParallelSubtasks()

接收运算符/函数并行度

https://ci.apache.org/projects/flink/flink-docs-master/api/java/org/apache/flink/api/common/functions/RuntimeContext.html

要获取 RuntimeContext，请使用 RichMapFunction 来实现 ParseData 并在 中调用 getRuntimeContext()打开()。

https://ci.apache.org/projects/flink/flink-docs-master/api/java/org/apache/flink/api/common/functions/RichFunction.html

像这样(只显示相关方法):

class ParseDate extends RichMapFunction {
    private long parallelism;
    private long idCounter;

    public void open(Configuration parameters) {
        RuntimeContext ctx = getRuntimeContext();
        parallelism = ctx.getNumberOfParallelSubtasks();
        idCounter = ctx.getIndexOfThisSubtask();
    }

    public OutputDataType map(InputDataType value) {
        OutputDataType output = new OutputDataType();
        output.setID(idCounter);
        idCounter += parallelism;
        // further processing
        return output;
    }
}