在 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