scala - Apache Flink 上的 zipWithIndex

Question

我想为我的输入的每一行分配一个 id - 这应该是来自 0 的数字至 N - 1 ，其中 N是输入中的行数。

粗略地说，我希望能够执行以下操作:

val data = sc.textFile(textFilePath, numPartitions)
val rdd = data.map(line => process(line))
val rddMatrixLike = rdd.zipWithIndex.map { case (v, idx) => someStuffWithIndex(idx, v) }

但是在 Apache Flink 中。是否可以？

Answer 1

这现在是 Apache Flink 0.10-SNAPSHOT 版本的一部分。 zipWithIndex(in) 的示例和 zipWithUniqueId(in)可在官方Flink documentation .