scala - 如何在 Apache Flink 中获取 Partitioner？

Question

我们正在尝试为 Apache Flink 创建一个扩展，它使用自定义分区。对于某些运算符(operator)，我们想要检查/检索使用的分区器。不幸的是，我找不到在给定数据集上执行此操作的任何可能性。我错过了什么还是有其他解决方法？

我会从这样的事情开始:

class MyPartitioner[..](..) extends Partitioner[..] {..}
[..]
val myP = new MyPartitioner(...)
val ds = in.partitionCustom(myP, 0)

现在，我想从另一个类访问分区器(如果已定义)。在 Spark 中，我会这样做:

val myP = ds.partitioner.get.asInstanceOf[MyPartitioner]

但是，对于 Flink，我找不到这种可能性。

编辑1:

suggestion of Fabian 似乎有可能.但是，有两个限制:

(1) 使用 Scala 时，您必须先检索底层 Java DataSet 才能将其转换为 PartitionOperator
(2) 分区必须是最后的操作。因此在设置和获取分区器之间不能使用其他操作。例如。以下是不可能的:

val in: DataSet[(String, Int)] = ???

val myP = new MyPartitioner()
val ds = in.partitionCustom(myP, 0)
val ds2 = ds.map(x => x)

val myP2 = ds2.asInstanceOf[PartitionOperator].getCustomPartitioner

感谢你并致以真诚的问候，
菲利普

Answer 1

可以投返回的DataSet变成 PartitionOperator并调用PartitionOperator.getCustomPartitioner() :

val in: DataSet[(String, Int)] = ???

val myP = new MyPartitioner()
val ds = in.partitionCustom(myP, 0)

val myP2 = ds.asInstanceOf[PartitionOperator].getCustomPartitioner

注意

getCustomPartitioner()是一种内部方法(即，不是公共(public) API 的一部分)，可能会在 Flink 的 future 版本中发生变化。

PartitionOperator也用于其他分区类型，例如 DataSet.partitionByHash() .在这些情况下 getCustomPartitioner()可能返回 null .