flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Mich Talebzadeh <mich.talebza...@gmail.com>
Subject Re: Working out through individual messages in Flink
Date Tue, 07 Aug 2018 22:38:58 GMT
Hi Jorn,

Thanks I uploaded the Scala code to my GitHub  --> md_streaming.scala

https://github.com/michTalebzadeh/Flink

Regards,

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 Tue, 7 Aug 2018 at 23:29, Jörn Franke <jornfranke@gmail.com> wrote:

> Hi Mich,
>
> Would it be possible to share the full source code ?
> I am missing a call to streamExecEnvironment.execute
>
> Best regards
>
> On 8. Aug 2018, at 00:02, Mich Talebzadeh <mich.talebzadeh@gmail.com>
> wrote:
>
> Hi Fabian,
>
> Reading your notes above I have converted the table back to DataStream.
>
>     val tableEnv = TableEnvironment.getTableEnvironment(streamExecEnv)
>     tableEnv.registerDataStream("priceTable", splitStream, 'key, 'ticker,
> 'timeissued, 'price)
>
>            val key =
> tableEnv.scan("priceTable").select('key).toDataStream[Row]
>            val ticker =
> tableEnv.scan("priceTable").select('ticker).toDataStream[Row]
>            val timeissued =
> tableEnv.scan("priceTable").select('timeissued).toDataStream[Row]
>            val price =
> tableEnv.scan("priceTable").select('price).toDataStream[Row]
>
> My intension is to create an Hbase sink as follows:
>
>             // Save prices to Hbase table
>              var p = new Put(new String(key).getBytes())
>              p.add("PRICE_INFO".getBytes(), "TICKER".getBytes(),
> new String(ticker).getBytes())
>              p.add("PRICE_INFO".getBytes(), "ISSUED".getBytes(),     new
> String(timeissued).getBytes())
>              p.add("PRICE_INFO".getBytes(), "PRICE".getBytes(),
> new String(priceToString).getBytes())
>              p.add("PRICE_INFO".getBytes(), "CURRENCY".getBytes(),
> new String(CURRENCY).getBytes())
>              p.add("OPERATION".getBytes(), "OP_TYPE".getBytes(),
> new String(1.toString).getBytes())
>              p.add("OPERATION".getBytes(), "OP_TIME".getBytes(),
> new String(System.currentTimeMillis.toString).getBytes())
>              HbaseTable.put(p)
>              HbaseTable.flushCommits()
>
> However, I don't seem to be able to get the correct values for the columns!
>
> 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 09:58, Fabian Hueske <fhueske@gmail.com> wrote:
>
>> A *Table*Source [1], is a special input connector for Flink's relational
>> APIs (Table API and SQL) [2].
>> You can transform and filter with these APIs as well (it's probably even
>> easier). In SQL this would be the SELECT and WHERE clauses of a query.
>>
>> However, there is no *Table*Sink for HBase and you would need to convert
>> the Table back to a DataStream [3].
>> That's not very difficult since the APIs are integrated with each other.
>>
>> Best, Fabian
>>
>> [1]
>> https://ci.apache.org/projects/flink/flink-docs-release-1.5/dev/table/sourceSinks.html
>> [2]
>> https://ci.apache.org/projects/flink/flink-docs-release-1.5/dev/table/common.html
>> [3]
>> https://ci.apache.org/projects/flink/flink-docs-release-1.5/dev/table/common.html#convert-a-table-into-a-datastream-or-dataset
>>
>> 2018-07-30 10:47 GMT+02:00 Mich Talebzadeh <mich.talebzadeh@gmail.com>:
>>
>>> Thanks Fabian. That was very useful.
>>>
>>> How about an operation like below?
>>>
>>>          // create builder
>>>          val KafkaTableSource = Kafka011JsonTableSource.builder()
>>>          // set Kafka topic
>>>             .forTopic(topicsValue)
>>>          // set Kafka consumer properties
>>>             .withKafkaProperties(properties)
>>>          // set Table schema
>>>         .withSchema(TableSchema.builder()
>>>         .field("key", Types.STRING)
>>>         .field("ticker", Types.STRING)
>>>         .field("timeissued", Types.STRING)
>>>         .field("price", Types.FLOAT)
>>>         .build())
>>>
>>> Will that be OK?
>>>
>>>
>>> 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 09:19, Fabian Hueske <fhueske@gmail.com> wrote:
>>>
>>>> Hi,
>>>>
>>>> Flink processes streams record by record, instead of micro-batching
>>>> records together. Since every record comes by itself, there is no for-each.
>>>> Simple record-by-record transformations can be done with a MapFunction,
>>>> filtering out records with a FilterFunction. You can also implement a
>>>> FlatMapFunction to do both in one step.
>>>>
>>>> Once the stream is transformed and filtered, you can write it to HBase
>>>> with a sink function.
>>>>
>>>>
>>>> 2018-07-30 10:03 GMT+02:00 Mich Talebzadeh <mich.talebzadeh@gmail.com>:
>>>>
>>>>> Just to clarify these are the individual prices separated by ','. The
>>>>> below shows three price lines in the topic
>>>>>
>>>>> UUID,                            Security,         Time,        Price
>>>>> 1230a9a9-dc57-40e4-a000-6ea6989c754a, MRW, 2018-07-28T20:38:43, 241.88
>>>>> 8143c6ca-109f-484f-be0e-9d0f5ca32f07,SAP,2018-07-28T20:38:44,56.94
>>>>> 81a54ff8-6ac8-470a-a522-51737d685264,VOD,2018-07-28T20:38:44,219.33
>>>>>
>>>>>
>>>>> 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 07:58, Mich Talebzadeh <
>>>>> mich.talebzadeh@gmail.com> wrote:
>>>>>
>>>>>>
>>>>>> Hi,
>>>>>>
>>>>>> I have a Kafka topic that transmits 100 security prices ever 2
>>>>>> seconds.
>>>>>>
>>>>>> In Spark streaming I go through the RDD and walk through rows one
by
>>>>>> one and check prices
>>>>>> In prices are valuable I store them into an Hbase table
>>>>>>
>>>>>>     val dstream = KafkaUtils.createDirectStream[String, String,
>>>>>> StringDecoder, StringDecoder](streamingContext, kafkaParams, topics)
>>>>>>     dstream.cache()
>>>>>>     dstream.foreachRDD
>>>>>>     { pricesRDD =>
>>>>>>           // Work on individual messages
>>>>>>       *   for(line <- pricesRDD.collect.toArray)*
>>>>>>          {
>>>>>>            var key = line._2.split(',').view(0).toString
>>>>>>            var ticker =  line._2.split(',').view(1).toString
>>>>>>            var timeissued = line._2.split(',').view(2).toString
>>>>>>            var price = line._2.split(',').view(3).toFloat
>>>>>>            val priceToString = line._2.split(',').view(3)
>>>>>>             if (price > 90.0)
>>>>>>            {
>>>>>>                //save to Hbase table
>>>>>>            }
>>>>>>           }
>>>>>>      }
>>>>>>
>>>>>> That works fine.
>>>>>>
>>>>>> In Flink I define my source as below
>>>>>>
>>>>>>     val streamExecEnv =
>>>>>> StreamExecutionEnvironment.getExecutionEnvironment
>>>>>>
>>>>>> streamExecEnv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
>>>>>>     val stream = streamExecEnv
>>>>>>        .addSource(new FlinkKafkaConsumer011[String](topicsValue,
new
>>>>>> SimpleStringSchema(), properties))
>>>>>>
>>>>>> Is there anyway I can perform similar operation in Flink? I need
to
>>>>>> go through every topic load sent and look at the individual rows/
For
>>>>>> example what is the equivalent of
>>>>>>
>>>>>> *for(line <- pricesRDD.collect.toArray)*
>>>>>> In flink?
>>>>>>
>>>>>> 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.
>>>>>>
>>>>>>
>>>>>>
>>>>>
>>>>
>>

Mime
View raw message