命名 Flink 运算符

Question

我正在使用Flink v.1.4.0.

我希望能够在Flink UI中命名运算符。我知道要做到这一点，我所需要的只是在DataSet或DataStream上使用.name()方法。例如，像这样：

DataSet<SomePojo> flinkDataSet = ...;
flinkDataSet
.map(new SomeTransformation())
.name("Transformation A");

我想知道的是运算符的确切定义是什么？它是 DataSet/DataStream 的键吗？或者也许是元素类型？我知道如果我应用与应用函数的元素具有相同返回类型的MapFunction，运算符将保持不变，但请注意，如果我应用FlatMapFunction或KeyBy或GroupBy操作，运算符通常会更改，以及应用以下操作， 课程。

我问这个问题的原因是，我想了解我可以在 Flink UI 中让 Flink DAG 有多细化，例如，我可以命名每个转换吗？或者运算符是由正在处理的数据集/数据流的类型定义的，因此我只能命名它们？例如，这行得通吗？

flinkDataSet
.map(new SomeTransformation())
.name("Transformation A")
.map(new SomeOtherTransformation())
.name("Transformation B");

还是第一个名字会被第二个名字覆盖？我当然可以尝试并找出答案，但我也想了解原因。

Answer 1

就我使用 name() 函数而言，您可以使用数据流或数据集 API 执行的每个变形都是一个操作，因此在第二种情况下，图形将如下所示

"Transformation A" -------> "Transformation B"

但是在 Flink dashborad 上，这些任务有可能被分组到同一个任务中，如下所示：

-----------------------         -----------------------
| SOURCE              |  --->   |"Transformation A" ->| ---> More operations
-----------------------         |"Transformation B"   |
-----------------------

Answer 2

是的，您可以命名每个转换。但正如@diegoreico所指出的，如果运算符是链接的*，那么 Flink UI 将显示它们分组到 DAG 中的一个框(节点)中。

*链式运算符：参见 Flink docu 任务链和资源组