У кого-нибудь есть пример обобщенной c ProcessFunction во Flink?

Question

Под "generi c" я имею в виду возможность принимать любой тип объекта в качестве ввода и возвращать тот же объект в качестве вывода.

Допустим, задача функции состоит в том, чтобы сериализовать каждый элемент в json и записать его в качестве побочного вывода.

class MyProcessFunction() extends ProcessFunction[? , ?] {

    def processElement(element: ?, ctx: ProcessFunction[?, ?]#Context, out: Collector[?]): Unit = ??? 

    ... 
}

Могу ли я определить это так, чтобы оно могло использоваться различными типами входов?

Answer 1

Вы можете сделать это, сделав Ваш класс обобщенным c. Таким образом, у вас будет что-то вроде:

class MyProcessFunction[T] extends ProcessFunction[T, T] {
  override def processElement(value: T, ctx: ProcessFunction[T, T]#Context, out: Collector[T]): Unit = ???
}

Таким образом, вы сможете определить тип, когда создадите экземпляр функции.