如何将 Flink Source [ Data Stream ] String 值转换为 Map？答案

【问题标题】：How to transform Flink Source [ Data Stream ] String value to Map?如何将 Flink Source [ Data Stream ] String 值转换为 Map？
【发布时间】：2019-05-17 15:25:09
【问题描述】：

尝试使用 scala 将 Apache Flink 源字符串转换为 Map。

我的源流字符串：key1=value1key2=2000-12-17 00:00:00key3=Testkey4=08.89198key5=103.000

代码：

val environment = StreamExecutionEnvironment.getExecutionEnvironment
val out  = environment.addSource(...)
val mapper = new ObjectMapper()
val texToMap = out.map(mapper.readValue(_,classOf[Map[Object,Object]])
println(textToJson)

它的抛出错误就像

org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonParseException: 无法识别的令牌 'key1': 期待 ('true', 'false' 或 'null')

由于我是 flink 实现的新手，有什么方法可以将 String 值转换为 Map。

【问题讨论】：

标签： json scala apache-flink flink-streaming data-stream

【解决方案1】：

Apache Flink 使用 jackson shaded jar 用于一些特殊用途。在您的情况下，您需要导入 com.fasterxml.jackson.databind.ObjectMapper 而不是 Flink 的阴影杰克逊依赖项。
同时，您需要这个，因为您使用的是 Scala。

import com.fasterxml.jackson.module.scala.DefaultScalaModule
val mapper = new ObjectMapper()
mapper.registerModule(DefaultScalaModule)

【讨论】：