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
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.
