Flink - 导入类型不能修复“找不到类型的证据参数的隐式值....TypeInformation”答案

【问题标题】：Flink - importing types doesn't fix "could not find implicit value for evidence parameter of type ....TypeInformation"Flink - 导入类型不能修复“找不到类型的证据参数的隐式值....TypeInformation”
【发布时间】：2021-05-01 22:18:45
【问题描述】：

在 Scala Flink 中，无论我尝试什么，都会不断收到如下错误：

找不到类型的证据参数的隐式值 org.apache.flink.api.common.typeinfo.TypeInformation[String] .map(t => t)

我已经尝试过很明显的导入：

    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.api.scala._

但这对编译错误没有帮助。我的目标是从字符串中解析 JSON 值，但是当我什至无法将字符串映射到字符串时（更不用说在映射中执行 parse(t)），我该怎么做？

我正在使用 Flink 1.12.1 和 Scala 2.12。

object AmplitudeExample {
  def main(args: Array[String]) {
    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.api.scala._
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val text = env.readTextFile("/Users/dbost/src/amplitude-flink/example-data.json")

    val partitionedEvents = text
      .map(t => t)
     
    partitionedEvents.print()
  }
}

如果我可以让它工作，那么我的下一个任务是用 circe 解析字符串，如下所示：

import io.circe.parser._

object AmplitudeExample {
  def main(args: Array[String]) {
    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.api.scala._
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val text = env.readTextFile("/Users/dbost/src/amplitude-flink/example-data.json")

    val partitionedEvents = text
      .map(t => parse(t))

    partitionedEvents.print()
  }
}

【问题讨论】：

标签： json scala apache-flink flink-streaming

【解决方案1】：

调用 map(...) 时，尝试像这样添加 TypeInformation map(...)(TypeInformation.of(classOf[String])

【讨论】：

【解决方案2】：

你可以试试这个。

object AmplitudeExample {
  def main(args: Array[String]) {
    import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, _}
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val text = env.readTextFile("/data/tpcds/test")

    val partitionedEvents = text
      .map(t => t)

    partitionedEvents.print()
  }
}

【讨论】：

你认为我弄错了 StreamExecutionEnvironment？