Flink - importing types doesn't fix "could not find implicit value for evidence parameter of type ....TypeInformation"

1
votes

In Scala Flink, no matter what I try, I keep getting an error like this:

could not find implicit value for evidence parameter of type org.apache.flink.api.common.typeinfo.TypeInformation[String] .map(t => t)

I've tried the obvious thing of importing:

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

but that didn't help the compilation error. My goal is to parse a JSON value from a string, but how can I do that when I can't even map a string to a string (let alone perform parse(t) in the map)?

I'm using Flink 1.12.1 and 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()
  }
}

If I can get that working, then my next task is to parse the string with circe, like this:

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()
  }
}
2
jsonscalaapache-flinkflink-streaming

2 Answers

0
votes

When calling map(...), try add TypeInformation like this map(...)(TypeInformation.of(classOf[String])

0
votes

You could try this.

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()
  }
}