scala - Runtime Context Links to wrong output type -




it seems processfunction class' context looking flink.util.outputtag output tag it's context.output() function instead of streaming.api.scala.outputtag. , datastream.getsideoutput() looking streaming.api.scala.outputtag

sample code of mine:

val outputtag  = outputtag[(string, long, long)]("output") //stuff .process {   new processfunction[(string, long), (string, long, long)] {     lazy val state: valuestate[event] = getruntimecontext.getstate(new valuestatedescriptor[event]("stats", classof[event]))        override def processelement(value: (string, long), ctx: processfunction[(string, long), (string, long, long)]#context, out: collector[(string, long, long)]) {         val current: event = state.value() match {           case null =>             event(value._1, 1, value._2)           case event(key, count, total) =>             event(key, count + 1, totaltos + value._2)         }          state.update(current)          ctx.output(outputtag, (current.key, current.count, current.total))       }     }   }       val  sideoutputstream: datastream[(string, long, long)] = maindatastream.getsideoutput(outputtag) 

the error:

error:(55, 24) type mismatch; found   : org.apache.flink.streaming.api.scala.outputtag[(string, long, long)] required: org.apache.flink.util.outputtag[(string, long, anyval)] note: (string, long, long) <: (string, long, anyval), java-defined class  outputtag invariant in type t. may wish investigate wildcard type such `_ <: (string, long,  anyval)`. (sls 3.2.10)     

i using flink 1.3.2, scala 2.11. there anyway flink side ouput example(found: https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/stream/side_output.html ) work correctly in scala using 1.3.2?





wiki

Comments

Popular posts from this blog

elasticsearch - what is the equivalent data type for geo_point in hibernate search? -

Jenkins: find build number for git commit -

firebase - How to wait value in Ionic 2 -