命名Flink运算符

Question

我正在使用Flink v.1.4.0。

我希望能够在Flink UI中命名运算符。我明白这样做我只需要在.name()或DataSet上使用DataStream方法。例如，像这样：

DataSet<SomePojo> flinkDataSet = ...;

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

我想知道的是究竟是什么定义了一个运营商？它是DataSet的/ DataStream的关键吗？或者元素类型？我理解，如果我应用与函数所应用的元素具有相同返回类型的MapFunction，则运算符将保持不变，但注意到如果我应用FlatMapFunction或KeyBy或GroupBy操作，操作符通常会更改当然，应用行动。

我问这个的原因是因为我想了解我可以让Flink DAG在Flink UI中的粒度，例如：我能说出每一次变革吗？或者是由正在处理的DataSet / DataStream的类型定义的运算符，因此我只能命名那些？例如，这会有用吗？

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

或者第一个名字会被第二个名字覆盖？我可以试验并找出当然，但我也想了解这些问题。

Answer 1

至于我使用了name（）函数，你可以使用数据流或数据集API进行的每个转换都是一个操作，所以在第二种情况下，图形看起来像这样

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

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

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

Answer 2

是的，您可以为每次转型命名。但是正如@diegoreico指出的那样，如果操作符是流水线的，那么Flink UI将显示它们分组到DAG中的单个框（节点）中。