在 Flink 中的两个不同流中使用相同的运算符

Question

我想在两个不同的流中使用相同的运算符。但是，我收到一个错误，指出 UID 或该运算符不是唯一的。

lazy val opt: DataStream[Foo] => DataStream[Buzz] = src => src.flatMap(new MyFlatMapFunc).uid("opt")
lazy val pipe1 = : DataStream[Foo] => DataStream[Buzz] = src => opt(src). // Do keyBy and other logic
lazy val pipe2 = : DataStream[Foo] => DataStream[Buzz] = src => opt(src). // Do some other logic

我得到一个例外：

线程"main"java.lang.IllegalArgumentException中的异常：用户指定的ID"opt"上的哈希冲突。最可能的原因是非唯一 ID。请检查通过uid(String)指定的所有ID是否唯一。

Answer 1

这是因为uid位于将在管道中使用两次的运算符上。您有两个选项，您可以将两个流union为一个，以便运算符仅使用一次，或者您可以稍微更改逻辑，以便分配不同的 id：

lazy val opt: (DataStream[Foo], String) => DataStream[Buzz] = (src, id) => src.flatMap(new MyFlatMapFunc).uid(id)
lazy val pipe1 = : (DataStream[Foo], String) => DataStream[Buzz] = src => opt(src, "firstOpt"). // Do keyBy and other logic
lazy val pipe2 = : (DataStream[Foo], String) => DataStream[Buzz] = src => opt(src, "secondOpt"). // Do keyBy and other logic