Flink为单个源使用多个数据类

Question

一些代码：

implicit val formats = Serialization.formats(NoTypeHints)

case class DataClass(id: String, name: String)    

val dataSource = env
      .addSource(new FlinkKinesisConsumer[String](s"data-stream-$stage", new SimpleStringSchema, consumerConfig))
      .uid(s"data-stream-$stage-source-id").name("dataSource")
      .map(json => read[DataClass](json))

这里，我从kinesis流中获取数据，并将其序列化到我的数据类中。一切正常，但现在需要增加以另一种格式接收数据的能力（例如，DataClassSecond）

其中一个选项是，添加一个额外的数据源并在您自己的流中处理它们。

但是这需要一个额外的运动队列。我不确定这是否是一个好方法有没有什么方法可以从运动接收不同的数据，然后根据类型分割流？

Answer 1

您可以尝试根据字段过滤数据流，这样您将得到两个或多个仅包含正确JSON格式元素的流。

因此，最简单的方法是：

val streamDataClass = sourceStream.filter(_.contains("name"))
val streamDataClassSecond = sourceStream.filter(_.contains("surname"))

只有当name和surname对每个DataClass都是唯一的时，这才会起作用。更有效的做法可能是首先将映射DataStream到某种通用格式，或者使用要么之类的东西作为反序列化结果，然后检查它是否成功。