Hi,
could you please post the stacktrace with the exception and also let us
know which Flink version you're using?
I have tried the following code and it works on
master/flink-1.11/flink-1.10:
case class Foo(lang: String, count: Int)
def main(args: Array[String]): Unit = {
val senv = StreamExecutionEnvironment.getExecutionEnvironment
val stenv = StreamTableEnvironment.create(senv)
val source = senv.fromElements("Hello", "ciao")
val mapped = source.map( e => {Foo(e, 13) } )
stenv.registerDataStream("foo", mapped)
senv.execute()
}
It's not exactly your code but pretty similar and I use the same case class.
Best,
Aljoscha
On 09.07.20 22:44, Georg Heiler wrote:
> Hi,
>
> why can't I register the stream as a table and get a MalformedClassName
> exception?
>
> val serializer = new JSONKeyValueDeserializationSchema(false)
> val stream = senv.addSource(
> new FlinkKafkaConsumer(
> "tweets-raw-json",
> serializer,
> properties
> ).setStartFromEarliest() // TODO experiment with different start values
> )
>
> case class Foo(lang: String, count: Int)
> val r = stream
> .map(e => {
> Foo(e.get("value").get("lang").asText(), 1)
> })
> .keyBy(_.lang)
> .timeWindow(Time.seconds(10))
> .sum("count")
> r.print()
> stenv.registerDataStream("tweets_json", r)
>
> Best,
> Georg
>
Locked
