splitting DataStream throws error

[Mich Talebzadeh]

Hi,

I have Kafka streaming feeds where a row looks like below where fields are separated by ","

I can split them easily with split function

scala> val oneline = "05521df6-4ccf-4b2f-b874-eb27d461b305,IBM,2018-07-30T19:51:50,190.48"
oneline: String = 05521df6-4ccf-4b2f-b874-eb27d461b305,IBM,2018-07-30T19:51:50,190.48

scala> oneline.split(",")
res26: Array[String] = Array(05521df6-4ccf-4b2f-b874-eb27d461b305, IBM, 2018-07-30T19:51:50, 190.48)

I can get the individual columns as below

scala>            val key = oneline.split(",").map(_.trim).view(0).toString
key: String = 05521df6-4ccf-4b2f-b874-eb27d461b305

scala>            val key = oneline.split(",").map(_.trim).view(1).toString
key: String = IBM

scala>            val key = oneline.split(",").map(_.trim).view(2).toString
key: String = 2018-07-30T19:51:50

scala>            val key = oneline.split(",").map(_.trim).view(3).toFloat

key: Float = 190.48

Now when I apply the same to dataStream in flink it fails

val dataStream =  streamExecEnv

       .addSource(new FlinkKafkaConsumer011[String](topicsValue, new SimpleStringSchema(), properties))

dataStream.split(",")

[error] /home/hduser/dba/bin/flink/md_streaming/src/main/scala/myPackage/md_streaming.scala:154: type mismatch;
[error]  found   : String(",")
[error]  required: org.apache.flink.streaming.api.collector.selector.OutputSelector[String]
[error] dataStream.split(",")
[error]                  ^
[error] one error found
[error] (compile:compileIncremental) Compilation failed

What operation do I need to do on dataStream to make this split work?

Thanks

[Chesnay Schepler]

You define a flatMap function that takes a string, calls String#split on it and collects the array.

On 30.07.2018 22:04, Mich Talebzadeh wrote:

Hi,

I have Kafka streaming feeds where a row looks like below where fields are separated by ","

I can split them easily with split function

scala> val oneline = "05521df6-4ccf-4b2f-b874-eb27d461b305,IBM,2018-07-30T19:51:50,190.48"
oneline: String = 05521df6-4ccf-4b2f-b874-eb27d461b305,IBM,2018-07-30T19:51:50,190.48

scala> oneline.split(",")
res26: Array[String] = Array(05521df6-4ccf-4b2f-b874-eb27d461b305, IBM, 2018-07-30T19:51:50, 190.48)

I can get the individual columns as below

scala>            val key = oneline.split(",").map(_.trim).view(0).toString
key: String = 05521df6-4ccf-4b2f-b874-eb27d461b305

scala>            val key = oneline.split(",").map(_.trim).view(1).toString
key: String = IBM

scala>            val key = oneline.split(",").map(_.trim).view(2).toString
key: String = 2018-07-30T19:51:50

scala>            val key = oneline.split(",").map(_.trim).view(3).toFloat

key: Float = 190.48

Now when I apply the same to dataStream in flink it fails

val dataStream =  streamExecEnv

       .addSource(new FlinkKafkaConsumer011[String](topicsValue, new SimpleStringSchema(), properties))

dataStream.split(",")

[error] /home/hduser/dba/bin/flink/md_streaming/src/main/scala/myPackage/md_streaming.scala:154: type mismatch;
[error]  found   : String(",")
[error]  required: org.apache.flink.streaming.api.collector.selector.OutputSelector[String]
[error] dataStream.split(",")
[error]                  ^
[error] one error found
[error] (compile:compileIncremental) Compilation failed

What operation do I need to do on dataStream to make this split work?

Thanks

[Mich Talebzadeh]

Thanks

So the assumption is that one cannot perform split on DataStream[String] directly?

On Mon, 30 Jul 2018 at 21:54, Chesnay Schepler <[hidden email]> wrote:

You define a flatMap function that takes a string, calls String#split on it and collects the array.

On 30.07.2018 22:04, Mich Talebzadeh wrote:

Hi,

I have Kafka streaming feeds where a row looks like below where fields are separated by ","

I can split them easily with split function

scala> val oneline = "05521df6-4ccf-4b2f-b874-eb27d461b305,IBM,2018-07-30T19:51:50,190.48"
oneline: String = 05521df6-4ccf-4b2f-b874-eb27d461b305,IBM,2018-07-30T19:51:50,190.48

scala> oneline.split(",")
res26: Array[String] = Array(05521df6-4ccf-4b2f-b874-eb27d461b305, IBM, 2018-07-30T19:51:50, 190.48)

I can get the individual columns as below

scala>            val key = oneline.split(",").map(_.trim).view(0).toString
key: String = 05521df6-4ccf-4b2f-b874-eb27d461b305

scala>            val key = oneline.split(",").map(_.trim).view(1).toString
key: String = IBM

scala>            val key = oneline.split(",").map(_.trim).view(2).toString
key: String = 2018-07-30T19:51:50

scala>            val key = oneline.split(",").map(_.trim).view(3).toFloat

key: Float = 190.48

Now when I apply the same to dataStream in flink it fails

val dataStream =  streamExecEnv

       .addSource(new FlinkKafkaConsumer011[String](topicsValue, new SimpleStringSchema(), properties))

dataStream.split(",")

[error] /home/hduser/dba/bin/flink/md_streaming/src/main/scala/myPackage/md_streaming.scala:154: type mismatch;
[error]  found   : String(",")
[error]  required: org.apache.flink.streaming.api.collector.selector.OutputSelector[String]
[error] dataStream.split(",")
[error]                  ^
[error] one error found
[error] (compile:compileIncremental) Compilation failed

What operation do I need to do on dataStream to make this split work?

Thanks

[vino yang]

Answered Mich privately, copy here:

Hi Mich,

The use of Split directly on the stream object is wrong. 

It is used to split the data in the stream object, not the format of the stream object data itself. In this scenario, if you want to parse the data, use the map function only after the source stream object, and parse each piece of data in it. 

Of course, you can essentially customize the SourceFunction and parse it directly when you consume it, but you have already used the Kafka Consumer provided by Flink, which does not provide this functionality. 

So I suggest you do it with MapFunction.

Thanks, vino.

2018-07-31 5:21 GMT+08:00 Mich Talebzadeh <[hidden email]>:

Thanks

So the assumption is that one cannot perform split on DataStream[String] directly?

On Mon, 30 Jul 2018 at 21:54, Chesnay Schepler <[hidden email]> wrote:

You define a flatMap function that takes a string, calls String#split on it and collects the array.

On 30.07.2018 22:04, Mich Talebzadeh wrote:

Hi,

I have Kafka streaming feeds where a row looks like below where fields are separated by ","

I can split them easily with split function

scala> val oneline = "05521df6-4ccf-4b2f-b874-eb27d461b305,IBM,2018-07-30T19:51:50,190.48"
oneline: String = 05521df6-4ccf-4b2f-b874-eb27d461b305,IBM,2018-07-30T19:51:50,190.48

scala> oneline.split(",")
res26: Array[String] = Array(05521df6-4ccf-4b2f-b874-eb27d461b305, IBM, 2018-07-30T19:51:50, 190.48)

I can get the individual columns as below

scala>            val key = oneline.split(",").map(_.trim).view(0).toString
key: String = 05521df6-4ccf-4b2f-b874-eb27d461b305

scala>            val key = oneline.split(",").map(_.trim).view(1).toString
key: String = IBM

scala>            val key = oneline.split(",").map(_.trim).view(2).toString
key: String = 2018-07-30T19:51:50

scala>            val key = oneline.split(",").map(_.trim).view(3).toFloat

key: Float = 190.48

Now when I apply the same to dataStream in flink it fails

val dataStream =  streamExecEnv

       .addSource(new FlinkKafkaConsumer011[String](topicsValue, new SimpleStringSchema(), properties))

dataStream.split(",")

[error] /home/hduser/dba/bin/flink/md_streaming/src/main/scala/myPackage/md_streaming.scala:154: type mismatch;
[error]  found   : String(",")
[error]  required: org.apache.flink.streaming.api.collector.selector.OutputSelector[String]
[error] dataStream.split(",")
[error]                  ^
[error] one error found
[error] (compile:compileIncremental) Compilation failed

What operation do I need to do on dataStream to make this split work?

Thanks