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 <mich.talebza...@gmail.com>: > Thanks > > So the assumption is that one cannot perform split on DataStream[String] > directly? > > Dr Mich Talebzadeh > > > > LinkedIn * > https://www.linkedin.com/profile/view?id=AAEAAAAWh2gBxianrbJd6zP6AcPCCdOABUrV8Pw > <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 Mon, 30 Jul 2018 at 21:54, Chesnay Schepler <ches...@apache.org> 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 >> >> Dr Mich Talebzadeh >> >> >> >> LinkedIn * >> https://www.linkedin.com/profile/view?id=AAEAAAAWh2gBxianrbJd6zP6AcPCCdOABUrV8Pw >> <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. >> >> >> >> >>
