Flink动态表和流连接

Question

我试图从动态表和基于某些字段的流中派生新表。

有没有人能为你提供最好的指导。我对flink和尝试新事物是陌生的。

//Dynamic Table

Table books = tEnv.sqlQuery("SELECT bookId, instrument, sum(tradedQuanity) as totalQuantity FROM tradeStreamTable group by bookId, instrument");

tEnv.registerTable("books", books);

书籍

============================
BookId， Instruments， Quantity
Book1， Goog，100
Book2， Vod，10
Book1， Appl，50
Book2， Goog，60
Book1， Vod，130
Book3， Appl，110

//My Stream

tEnv.registerDataStream("allInstrumentsTable", allInstruments, "timeStampMs, instrument, instrumentValue ");

allInstrumentsTable公司

====================================================================================================“时间戳、工具、工具值（价格）
流。。。。。。

=========================================

每当我在book table中得到新的更改或流中的instrumentValue得到新的instrumentValue时，我都试图派生新的表（动态）。连接仪表，仪表值*总数量。

预订-最新价格（新表格）

======================================
BookId， Instruments， Quantity， InstrentValue*总计数量
Book1， Goog，100,1203
Book1， Appl，50，...
Book1， Vod，130，...
Book2， Vod，10，...
Book2， Goog，60，...
Book3， Appl，110，...

Answer 1

您不能从外部“更新”表。流中的表类似于RDBMS中的物化视图。它们是从特定时刻的流状态导出的视图。

你能做的就是从这两个表中得到一个新表

SELECT instrument, instrumentValue * totalQuantity FROM allInstrumentsTable aJOIN books b ON a.instrument = b.instrument;

由于这是非窗口连接，您还应该考虑一些保留策略，以确保状态不会无限期增长。