osdir.com


[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]

Re: Converting a DataStream into a Table throws error


Hi Mich,

I have reviewed your code in the github you provided.

I copied your code to org.apache.flink.table.examples.scala under flink-examples-table. It passed the compilation and didn't report the exception you provided, although there are other exceptions (it's about hdfs, this is because of my environment).
First of all, the necessary dependencies are just a few:

<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table_${scala.binary.version}</artifactId>
<version>${project.version}</version>
</dependency>

<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
<version>${project.version}</version>
</dependency>

<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka-0.11_${scala.binary.version}</artifactId>
<version>${project.version}</version>
</dependency>

Please remove irrelevant dependencies.

In addition, your StreamExecutionEnvironment object has a problem with its reference. The correct path should be:
Import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

In addition, you specify four fields, but your stream object is DataStream[String]. This is also a mistake. You are advised to parse it first with a MapFunction. The example is as follows:

    Val dataStream = streamExecEnv
      .addSource(new FlinkKafkaConsumer011[String](topicsValue, new SimpleStringSchema(), properties))

    Val newDateStream = dataStream.map(new MapFunction[String, Tuple4[String, String, String, String]] {
      Override def map(value: String): Tuple4[String, String, String, String] = {
        Return null
      }
    })

    Val tableEnv = TableEnvironment.getTableEnvironment(streamExecEnv)
    //tableEnv.registerDataStream("table1", streamExecEnv, 'key, 'ticker, 'timeissued, 'price)
    tableEnv.registerDataStream("priceTable", newDateStream, 'key, 'ticker, 'timeissued, 'price)

Thanks, vino.

2018-08-02 20:36 GMT+08:00 Mich Talebzadeh <mich.talebzadeh@xxxxxxxxx>:
Appreciate if anyone had a chance to look at the Scala code in GitHub and advise


Regards,

Dr Mich Talebzadeh

 

LinkedIn  https://www.linkedin.com/profile/view?id=AAEAAAAWh2gBxianrbJd6zP6AcPCCdOABUrV8Pw

 

http://talebzadehmich.wordpress.com


Disclaimer: Use it at your own risk. Any and all responsibility for any loss, damage or destruction of data or any other property which may arise from relying on this email's technical content is explicitly disclaimed. The author will in no case be liable for any monetary damages arising from such loss, damage or destruction.

 



On Thu, 2 Aug 2018 at 09:06, Mich Talebzadeh <mich.talebzadeh@xxxxxxxxx> wrote:
Thanks Timo,

Did as suggested getting this compilation error

[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:136: could not find implicit value for evidence parameter of type org.apache.flink.api.common.typeinfo.TypeInformation[String]
[error]        .addSource(new FlinkKafkaConsumer011[String](topicsValue, new SimpleStringSchema(), properties))
[error]                  ^
[error] one error found
[error] (compile:compileIncremental) Compilation failed

Dr Mich Talebzadeh

 

LinkedIn  https://www.linkedin.com/profile/view?id=AAEAAAAWh2gBxianrbJd6zP6AcPCCdOABUrV8Pw

 

http://talebzadehmich.wordpress.com


Disclaimer: Use it at your own risk. Any and all responsibility for any loss, damage or destruction of data or any other property which may arise from relying on this email's technical content is explicitly disclaimed. The author will in no case be liable for any monetary damages arising from such loss, damage or destruction.

 



On Thu, 2 Aug 2018 at 09:01, Timo Walther <twalthr@xxxxxxxxxx> wrote:
Whenever you use Scala and there is a Scala specific class use it.

remove: import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
add: import org.apache.flink.streaming.api.scala._

This will use org.apache.flink.streaming.api.scala.StreamExecutionEnvironment.

Timo

Am 02.08.18 um 09:47 schrieb Mich Talebzadeh:
Tremendous. Many thanks.

Put the sbt build file and the Scala code here


Regards,

Dr Mich Talebzadeh

 

LinkedIn  https://www.linkedin.com/profile/view?id=AAEAAAAWh2gBxianrbJd6zP6AcPCCdOABUrV8Pw

 

http://talebzadehmich.wordpress.com


Disclaimer: Use it at your own risk. Any and all responsibility for any loss, damage or destruction of data or any other property which may arise from relying on this email's technical content is explicitly disclaimed. The author will in no case be liable for any monetary damages arising from such loss, damage or destruction.

 



On Thu, 2 Aug 2018 at 08:27, Timo Walther <twalthr@xxxxxxxxxx> wrote:
Hi Mich,

could you share your project with us (maybe on github)? Then we can import it and debug what the problem is.

Regards,
Timo

Am 02.08.18 um 07:37 schrieb Mich Talebzadeh:
Hi Jorn,

Here you go the dependencies

libraryDependencies += "org.apache.hadoop" % "hadoop-core" % "1.2.1"
libraryDependencies += "org.apache.hbase" % "hbase" % "1.2.6"
libraryDependencies += "org.apache.hbase" % "hbase-client" % "1.2.6"
libraryDependencies += "org.apache.hbase" % "hbase-common" % "1.2.6"
libraryDependencies += "org.apache.hbase" % "hbase-server" % "1.2.6"
libraryDependencies += "org.apache.flink" %% "flink-connector-kafka-0.11" % "1.5.0"
libraryDependencies += "org.apache.flink" %% "flink-connector-kafka-base" % "1.5.0"
libraryDependencies += "org.apache.flink" %% "flink-scala" % "1.5.0"
libraryDependencies += "org.apache.kafka" % "kafka-clients" % "0.11.0.0"
libraryDependencies += "org.apache.flink" %% "flink-streaming-scala" % "1.5.0"
libraryDependencies += "org.apache.flink" %% "flink-table" % "1.5.0" % "provided"
libraryDependencies += "org.apache.kafka" %% "kafka" % "0.11.0.0"

Thanks

Dr Mich Talebzadeh

 

LinkedIn  https://www.linkedin.com/profile/view?id=AAEAAAAWh2gBxianrbJd6zP6AcPCCdOABUrV8Pw

 

http://talebzadehmich.wordpress.com


Disclaimer: Use it at your own risk. Any and all responsibility for any loss, damage or destruction of data or any other property which may arise from relying on this email's technical content is explicitly disclaimed. The author will in no case be liable for any monetary damages arising from such loss, damage or destruction.

 



On Thu, 2 Aug 2018 at 06:19, Jörn Franke <jornfranke@xxxxxxxxx> wrote:

How does your build.sbt looks especially dependencies?
On 2. Aug 2018, at 00:44, Mich Talebzadeh <mich.talebzadeh@xxxxxxxxx> wrote:

Changed as suggested

   val  streamExecEnv = StreamExecutionEnvironment.getExecutionEnvironment
     val dataStream =  streamExecEnv
       .addSource(new FlinkKafkaConsumer011[String](topicsValue, new SimpleStringSchema(), properties))
  val tableEnv = TableEnvironment.getTableEnvironment(streamExecEnv)

  tableEnv.registerDataStream("table1", streamExecEnv, 'key, 'ticker, 'timeissued, 'price)

Still the same error

[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:139: overloaded method value registerDataStream with alternatives:
[error]   [T](name: String, dataStream: org.apache.flink.streaming.api.datastream.DataStream[T], fields: String)Unit <and>
[error]   [T](name: String, dataStream: org.apache.flink.streaming.api.datastream.DataStream[T])Unit
[error]  cannot be applied to (String, org.apache.flink.streaming.api.environment.StreamExecutionEnvironment, Symbol, Symbol, Symbol, Symbol)
[error]   tableEnv.registerDataStream("table1", streamExecEnv, 'key, 'ticker, 'timeissued, 'price)
[error]            ^
[error] one error found
[error] (compile:compileIncremental) Compilation failed
[error] Total time: 3 s, completed Aug 1, 2018 11:40:47 PM

Thanks anyway.

Dr Mich Talebzadeh

 

LinkedIn  https://www.linkedin.com/profile/view?id=AAEAAAAWh2gBxianrbJd6zP6AcPCCdOABUrV8Pw

 

http://talebzadehmich.wordpress.com


Disclaimer: Use it at your own risk. Any and all responsibility for any loss, damage or destruction of data or any other property which may arise from relying on this email's technical content is explicitly disclaimed. The author will in no case be liable for any monetary damages arising from such loss, damage or destruction.

 



On Wed, 1 Aug 2018 at 23:34, Fabian Hueske <fhueske@xxxxxxxxx> wrote:
Hi,

You have to pass the StreamExecutionEnvironment to the getTableEnvironment() method, not the DataStream (or DataStreamSource).
Change

val tableEnv = TableEnvironment.getTableEnvironment(dataStream)

to

val tableEnv = TableEnvironment.getTableEnvironment(streamExecEnv)

Best,
Fabian

2018-08-02 0:10 GMT+02:00 Mich Talebzadeh <mich.talebzadeh@xxxxxxxxx>:
Hi,

FYI, these are my imports

import java.util.Properties
import java.util.Arrays
import org.apache.flink.api.common.functions.MapFunction
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala
import org.apache.flink.streaming.util.serialization.SimpleStringSchema
import org.apache.flink.table.api.TableEnvironment
import org.apache.flink.table.api.scala._
import org.apache.flink.api.scala._
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.clients.consumer.KafkaConsumer
import org.apache.flink.core.fs.FileSystem
import org.apache.flink.streaming.api.TimeCharacteristic
import org.slf4j.LoggerFactory
import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaConsumer011, FlinkKafkaProducer011}
import java.util.Calendar
import java.util.Date
import java.text.DateFormat
import java.text.SimpleDateFormat
import org.apache.log4j.Logger
import org.apache.log4j.Level
import sys.process.stringSeqToProcess
import java.io.File

And this is the simple code

    val properties = new Properties()
    properties.setProperty("bootstrap.servers", bootstrapServers)
    properties.setProperty("zookeeper.connect", zookeeperConnect)
    properties.setProperty("group.id", flinkAppName)
    properties.setProperty("auto.offset.reset", "latest")
    val  streamExecEnv = StreamExecutionEnvironment.getExecutionEnvironment
    val dataStream =  streamExecEnv
       .addSource(new FlinkKafkaConsumer011[String](topicsValue, new SimpleStringSchema(), properties))
  val tableEnv = TableEnvironment.getTableEnvironment(dataStream)

  tableEnv.registerDataStream("table1", dataStream, 'key, 'ticker, 'timeissued, 'price)

And this is the compilation error

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:138: overloaded method value getTableEnvironment with alternatives:
[error]   (executionEnvironment: org.apache.flink.streaming.api.scala.StreamExecutionEnvironment)org.apache.flink.table.api.scala.StreamTableEnvironment <and>
[error]   (executionEnvironment: org.apache.flink.streaming.api.environment.StreamExecutionEnvironment)org.apache.flink.table.api.java.StreamTableEnvironment <and>
[error]   (executionEnvironment: org.apache.flink.api.scala.ExecutionEnvironment)org.apache.flink.table.api.scala.BatchTableEnvironment <and>
[error]   (executionEnvironment: org.apache.flink.api.java.ExecutionEnvironment)org.apache.flink.table.api.java.BatchTableEnvironment
[error]  cannot be applied to (org.apache.flink.streaming.api.datastream.DataStreamSource[String])
[error]   val tableEnv = TableEnvironment.getTableEnvironment(dataStream)
[error]                                   ^
[error] one error found
[error] (compile:compileIncremental) Compilation failed
[error] Total time: 3 s, completed Aug 1, 2018 11:02:33 PM
Completed compiling


which is really strange

Dr Mich Talebzadeh

 

LinkedIn  https://www.linkedin.com/profile/view?id=AAEAAAAWh2gBxianrbJd6zP6AcPCCdOABUrV8Pw

 

http://talebzadehmich.wordpress.com


Disclaimer: Use it at your own risk. Any and all responsibility for any loss, damage or destruction of data or any other property which may arise from relying on this email's technical content is explicitly disclaimed. The author will in no case be liable for any monetary damages arising from such loss, damage or destruction.

 



On Wed, 1 Aug 2018 at 13:42, Fabian Hueske <fhueske@xxxxxxxxx> wrote:
Hi I think you are mixing Java and Scala dependencies.

org.apache.flink.streaming.api.datastream.DataStream is the DataStream of the Java DataStream API.
You should use the DataStream of the Scala DataStream API.

Best, Fabian

2018-08-01 14:01 GMT+02:00 Mich Talebzadeh <mich.talebzadeh@xxxxxxxxx>:
Hi,

I believed I tried Hequn's suggestion and tried again

import org.apache.flink.table.api.Table
import org.apache.flink.table.api.TableEnvironment
import org.apache.flink.table.api.scala._

Unfortunately I am still getting the same error!

[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:151: 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
[error] Total time: 3 s, completed Aug 1, 2018 12:59:44 PM
Completed compiling


Dr Mich Talebzadeh

 

LinkedIn  https://www.linkedin.com/profile/view?id=AAEAAAAWh2gBxianrbJd6zP6AcPCCdOABUrV8Pw

 

http://talebzadehmich.wordpress.com


Disclaimer: Use it at your own risk. Any and all responsibility for any loss, damage or destruction of data or any other property which may arise from relying on this email's technical content is explicitly disclaimed. The author will in no case be liable for any monetary damages arising from such loss, damage or destruction.

 



On Wed, 1 Aug 2018 at 10:03, Timo Walther <twalthr@xxxxxxxxxx> wrote:
If these two imports are the only imports that you added, then you did not follow Hequn's advice or the link that I sent you.

You need to add the underscore imports to let Scala do its magic.

Timo


Am 01.08.18 um 10:28 schrieb Mich Talebzadeh:
Hi Timo,

These are my two flink table related imports

import org.apache.flink.table.api.Table
import org.apache.flink.table.api.TableEnvironment

And these are my dependencies building with SBT

libraryDependencies += "org.apache.hadoop" % "hadoop-core" % "1.2.1"
libraryDependencies += "org.apache.hbase" % "hbase" % "1.2.6"
libraryDependencies += "org.apache.hbase" % "hbase-client" % "1.2.6"
libraryDependencies += "org.apache.hbase" % "hbase-common" % "1.2.6"
libraryDependencies += "org.apache.hbase" % "hbase-server" % "1.2.6"
libraryDependencies += "org.apache.flink" %% "flink-connector-kafka-0.11" % "1.5.0"
libraryDependencies += "org.apache.flink" %% "flink-connector-kafka-base" % "1.5.0"
libraryDependencies += "org.apache.flink" %% "flink-scala" % "1.5.0"
libraryDependencies += "org.apache.kafka" % "kafka-clients" % "0.11.0.0"
libraryDependencies += "org.apache.flink" %% "flink-streaming-java" % "1.5.0" % "provided"
libraryDependencies += "org.apache.flink" %% "flink-table" % "1.5.0" % "provided"
libraryDependencies += "org.apache.kafka" %% "kafka" % "0.11.0.0"

There appears to be conflict somewhere that cause this error

[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


Thanks


Dr Mich Talebzadeh

 

LinkedIn  https://www.linkedin.com/profile/view?id=AAEAAAAWh2gBxianrbJd6zP6AcPCCdOABUrV8Pw

 

http://talebzadehmich.wordpress.com


Disclaimer: Use it at your own risk. Any and all responsibility for any loss, damage or destruction of data or any other property which may arise from relying on this email's technical content is explicitly disclaimed. The author will in no case be liable for any monetary damages arising from such loss, damage or destruction.

 



On Wed, 1 Aug 2018 at 09:17, Timo Walther <twalthr@xxxxxxxxxx> wrote:
Hi Mich,

I would check you imports again [1]. This is a pure compiler issue that is unrelated to your actual data stream. Also check your project dependencies.

Regards,
Timo

[1] https://ci.apache.org/projects/flink/flink-docs-master/dev/table/common.html#implicit-conversion-for-scala

Am 01.08.18 um 09:30 schrieb Mich Talebzadeh:

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]