Re: Converting a DataStream into a Table throws error

Posted by Mich Talebzadeh on
URL: http://deprecated-apache-flink-user-mailing-list-archive.369.s1.nabble.com/Converting-a-DataStream-into-a-Table-throws-error-tp21975p21984.html

Hi both,

I added the import as Hequn suggested.

My stream is very simple and consists of 4 values separated by "," as below 

05521df6-4ccf-4b2f-b874-eb27d461b305,IBM,2018-07-30T19:51:50,190.48

So this is what I have been trying to do

Code

    val dataStream =  streamExecEnv
       .addSource(new FlinkKafkaConsumer011[String](topicsValue, new SimpleStringSchema(), properties))
 //
 //
  val tableEnv = TableEnvironment.getTableEnvironment(streamExecEnv)
  val table1: Table = tableEnv.fromDataStream(dataStream, 'key, 'ticker, 'timeissued, 'price)

note those four columns in Table1 definition

And this is the error being thrown

[info] Compiling 1 Scala source to /home/hduser/dba/bin/flink/md_streaming/target/scala-2.11/classes...
[error] /home/hduser/dba/bin/flink/md_streaming/src/main/scala/myPackage/md_streaming.scala:152: overloaded method value fromDataStream with alternatives:
[error]   [T](dataStream: org.apache.flink.streaming.api.datastream.DataStream[T], fields: String)org.apache.flink.table.api.Table <and>
[error]   [T](dataStream: org.apache.flink.streaming.api.datastream.DataStream[T])org.apache.flink.table.api.Table
[error]  cannot be applied to (org.apache.flink.streaming.api.datastream.DataStreamSource[String], Symbol, Symbol, Symbol, Symbol)
[error]   val table1: Table = tableEnv.fromDataStream(dataStream, 'key, 'ticker, 'timeissued, 'price)
[error]                                ^
[error] one error found
[error] (compile:compileIncremental) Compilation failed

I suspect dataStream may not be compatible with this operation?

Regards,

On Wed, 1 Aug 2018 at 04:51, Hequn Cheng <[hidden email]> wrote:

Hi, Mich

You can try adding "import org.apache.flink.table.api.scala._", so that the Symbol can be recognized as an Expression.

Best, Hequn 

On Wed, Aug 1, 2018 at 6:16 AM, Mich Talebzadeh <[hidden email]> wrote:

Hi,

I am following this example 

https://ci.apache.org/projects/flink/flink-docs-release-1.5/dev/table/common.html#integration-with-datastream-and-dataset-api

This is my dataStream which is built on a Kafka topic

   //
    //Create a Kafka consumer
    //
    val dataStream =  streamExecEnv
       .addSource(new FlinkKafkaConsumer011[String](topicsValue, new SimpleStringSchema(), properties))
 //
 //
  val tableEnv = TableEnvironment.getTableEnvironment(streamExecEnv)
  val table1: Table = tableEnv.fromDataStream(dataStream, 'key, 'ticker, 'timeissued, 'price)

While compiling it throws this error

[error] /home/hduser/dba/bin/flink/md_streaming/src/main/scala/myPackage/md_streaming.scala:169: overloaded method value fromDataStream with alternatives:
[error]   [T](dataStream: org.apache.flink.streaming.api.datastream.DataStream[T], fields: String)org.apache.flink.table.api.Table <and>
[error]   [T](dataStream: org.apache.flink.streaming.api.datastream.DataStream[T])org.apache.flink.table.api.Table
[error]  cannot be applied to (org.apache.flink.streaming.api.datastream.DataStreamSource[String], Symbol, Symbol, Symbol, Symbol)
[error]   val table1: Table = tableEnv.fromDataStream(dataStream, 'key, 'ticker, 'timeissued, 'price)
[error]                                ^
[error] one error found
[error] (compile:compileIncremental) Compilation failed

The topic is very simple, it is comma separated prices. I tried mapFunction and flatMap but neither worked! 

Thanks,