I need this operation to stored filtered rows in an Hbase table.
I can access an existing Hbase table through flink API
My challenge is to put rows into Hbase table. Something like below and I
don't seem to be able to extract individual column values from priceTable
* val key =
tableEnv.scan("priceTable").select('key).toDataStream[Row].print()
val ticker =
tableEnv.scan("priceTable").select('ticker).toDataStream[Row].print()
val timeissued =
tableEnv.scan("priceTable").select('timeissued).toDataStream[Row].print()
val price =
tableEnv.scan("priceTable").select('price).toDataStream[Row].print()*
val CURRENCY = "GBP"
val op_type = "1"
val op_time = System.currentTimeMillis.toString
/*
if (price > 99.0)
{
// 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()
if(tableEnv.scan("priceTable").filter('ticker == "VOD" &&
'price > 99.0))
{
sqltext = Calendar.getInstance.getTime.toString + ", Price
on "+ticker+" hit " +price.toString
//java.awt.Toolkit.getDefaultToolkit().beep()
println(sqltext)
}
}
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 17:07, Mich Talebzadeh <mich.talebza...@gmail.com>
wrote:
> Hi,
>
> The following works fine
>
> tableEnv.registerDataStream("priceTable", splitStream, 'key, 'ticker,
> 'timeissued, 'price)
> val result = tableEnv.scan("priceTable").filter('ticker === "VOD" &&
> 'price > 99.0).select('key, 'ticker, 'timeissued, 'price)
> val r = result.toDataStream[Row]
> r.print()
>
> Now I would like to get the individual column values from priceTable into
> local variables
>
> This does not seem to work
>
> 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]
>
> What alternatives are there?
>
> 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.
>
>
>
