JSON解码Base64数据流Apache Flink

Question

需要一些建议，我已经使用scala创建了一个flink作业来消费来自Kafka的消息。但是消息是用base64编码压缩的。我已经试过这个代码了

    val x_stream: DataStream[ObjectNode] = env
      .addSource(
        new FlinkKafkaConsumer010[ObjectNode](parameters.get("kafka.topic.source"),
          new JsonNodeDeserializationSchema(),
          kfk_props
        ).setStartFromEarliest()
      ).name("Topic Test")).rebalance

代码由于它不是有效的Json格式而失败。

然后我尝试使用SimpleStringSchema（），就像下面的代码一样

    val x_stream: DataStream[String] = env
      .addSource(
        new FlinkKafkaConsumer010[String](parameters.get("kafka.topic.source"),
          new SimpleStringSchema(),
          kfk_props
        ).setStartFromEarliest()
      ).name("Topic Test")).rebalance

Kafka的信息完美地消耗了，但是输出如下

.....
    Br?G"p)0?p?AF??g}?Ly?@?
    ??>??j?)??);?E?]<d╚? ?-?@?g?????'2???�?�o???r?z????Q$????p    F╔?7?yx+_'v?2???K&??O??c??D,c0F2??ny[?=??%?/?M1:???bq?yHt"A??5???

如何将此数据解码为有效的JSON？

此致敬意

Answer 1

您应该实现KafkaDeserializationSchema接口来定制解析逻辑。

首先，解码base64并将其转换为字节流，然后生成一个字符串对象，最后将其解析为json。