Ha, that's a neat workaround! Thanks for sharing Hequn! When doing this, you should however, ensure that all records with the same key arrive from the same input task to avoid inconsistent behavior due to records arriving out of order. This would be the case if you ingest the table directly from a Kafka topic that is partitioned by key.
Best, Fabian 2018-01-05 14:55 GMT+01:00 Hequn Cheng <chenghe...@gmail.com>: > hi Ghassan, > > TableSource in Flink doesn't support primary key now, but you can achieve > it by doing a group by manually. > > Such as: > val resultTable = tableEnv.sql( > " > SELECT ProductID, AVG(Rating) FROM > ( SELECT ReviewID, LAST_VALUE(ProductID), > LAST_VALUE(Approved) , LAST_VALUE(Rating) FROM Reviews GROUP BY ReviewID) > WHERE Approved = true GROUP BY ProductID > " > ) > > You have to implement the LAST_VALUE AggregateFunction. For implementation, > you can refer to the MAX AggregateFunction(MAX always return the max value > while LAST_VALUE always return the latest value). Also, you can find > documents about Aggregate Functions here: https://ci.apache.org/ > projects/flink/flink-docs-release-1.4/dev/table/udfs. > html#aggregation-functions > > Best, Hequn > > > 2018-01-05 18:31 GMT+08:00 Fabian Hueske <fhue...@gmail.com>: > > > Hi Ghassan, > > > > Flink's Table API / SQL does not support the upsert ingestion mode > > (updating table rows by key) yet but only append mode, i.e, each event of > > the data stream is appended to the table. > > Hence, it is not possible to implement your use case using SQL. > > > > An upsert ingestion mode will be added in future version of Flink. > > > > Best, Fabian > > > > 2017-12-21 5:20 GMT+01:00 Ghassan Yammine <Ghassan.Yammine@bazaarvoice. > com > > >: > > > > > Hi Timo, > > > > > > Thanks for your quick reply. > > > > > > I understand your statement about the SQL expression: It is grouped by > > > ProductID and performs the average computation over all the records it > > > sees. There is no concept of primary key and INSERT/UPDATE. > > > > > > Is there a way to do what I want using SQL without, or with, the > > > pre-processing of the stream before a SQL expression is executed? > > > > > > One thought I had was to transform the stream - prior to the SQL > > > expression - by doing something like morphing the value of the Rating > to > > be > > > relative (instead of absolute) to the prior Rating, for the same > review. > > > However, this would entail some review tracking mechanism. > > > > > > As I mentioned before, I’ve implemented this using the Datastream API - > > > internally maintaining a map of reviews and updating the average > > > calculation as each record is processed. Using SQL is very important > to > > > us, and I’d like to see if I can make it work. > > > > > > Regarding the output sink, I previously looked into that and concluded > > > that I could not perform any “adjustment” to the average calculation > > > because of what the current SQL expression emits (i.e. incorrect > average > > > calc.). Here is what the (retract) stream looks like post-SQL: > > > > > > (true,(product-100,1.0)) > > > (false,(product-100,1.0)) > > > (true,(product-100,2.5)) > > > (false,(product-100,2.5)) > > > (true,(product-100,3.6666666666666665)) > > > (false,(product-100,3.6666666666666665)) > > > (true,(product-100,3.5)). <———————— This is the correct value (per > the > > > SQL) but not what I want it to be. > > > > > > Even if I change what the SQL query returns/emits so that the TableSink > > > can perform the average calculation, the latter will have to track > prior > > > reviews in order to update its average calculation. If I’m correct, > then > > > this is essentially no different than the Datastream API implementation > > > that I have. > > > > > > Agains, thanks for your quick response and help. > > > > > > Regards, > > > > > > Ghassan > > > > > > > > > > > > On Dec 20, 2017, at 11:50 AM, Timo Walther <twal...@apache.org<mailto: > > twa > > > l...@apache.org>> wrote: > > > > > > Hi Ghassan, > > > > > > in your example the result 3.5 is correct. The query is executed with > > > standard SQL semantics. You only group by ProductID and since it is the > > > same for all elements, the average is 3.5. > > > > > > The second "review-3" does not replace anything. In general, the > > > replacement would happen in the TableSink. The dynamic table performs > > view > > > maintenance. The TableSink materializes the result to some key-value > > store > > > or database. > > > > > > It might be worth to look into TableSinks [0] and the JavaDocs of the > > > mentioned classes. > > > > > > Feel free to ask further questions if necessary. > > > > > > Regards, > > > Timo > > > > > > [0] https://ci.apache.org/projects/flink/flink-docs-release-1.4/ > > dev/table/ > > > sourceSinks.html#define-a-tablesink > > > > > > > > > > > > Am 12/19/17 um 9:08 PM schrieb Ghassan Yammine: > > > Hello, > > > > > > I’m knew to Flink and I need some help. I’d like to use the SQL API > for > > > processing an incoming stream that has the following characteristics: > > > > > > * Each stream record has a key > > > * The record can be updated > > > * The record is of the form: reviewId -> (productId, rating) > > > > > > For the above stream, I want to compute the average rating for each > > > product ID. The key is the reviewId > > > With the SQL API, I get incorrect results. However, I’ve been able to > > > make it work through the use of RichFlatMapFunction and the Datastream > > API. > > > > > > Below is the entire code listing, which does not work. I know I’m > > missing > > > the definition/use of a primary key so that an update on the same key > can > > > occur. > > > However, I’m not sure how to go about doing this. Any help/comments > are > > > welcome. > > > > > > Thank you, > > > > > > Ghassan > > > > > > > > > package com.bazaarvoice.flink_poc > > > > > > import com.bazaarvoice.flink_poc.flink_poc.{ProductId, ReviewId} > > > import org.apache.flink.api.common.time.Time > > > import org.apache.flink.streaming.api.TimeCharacteristic > > > import org.apache.flink.streaming.api.scala.{DataStream, > > > createTypeInformation, _} > > > import org.apache.flink.table.api.{StreamQueryConfig, > TableEnvironment} > > > import org.apache.flink.table.api.scala._ > > > > > > package object flink_poc{ > > > type ProductId = String > > > type ReviewId = String > > > } > > > > > > case class SummaryReview(reviewId: ReviewId, productId: ProductId, > > > approved: Boolean, rating: Double) extends Serializable { > > > override def toString: String = { > > > s"$reviewId, $productId, ${if (approved) "APPROVED" else > > > "REJECTED"}, $rating" > > > } > > > } > > > > > > object AverageRatingWithSQL { > > > > > > def main(args: Array[String]) { > > > > > > val events = List( > > > SummaryReview("review-1", "product-100", approved = true, 1), > > > SummaryReview("review-2", "product-100", approved = true, 4), > > > SummaryReview("review-3", "product-100", approved = true, 6), > > > SummaryReview("review-3", "product-100", approved = true, 3) > // > > > <-- this should override the previous record > > > ).toSeq > > > // Average rating should be equal to (1+4+3)/3 = 2.666667 > > > > > > val env = StreamExecutionEnvironment.getExecutionEnvironment > > > env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime) > > > > > > val inputStream: DataStream[SummaryReview] = > > env.fromCollection(events) > > > > > > val tableEnv = TableEnvironment.getTableEnvironment(env) > > > > > > tableEnv.registerDataStream("Reviews", inputStream, 'ReviewID, > > > 'ProductID, 'Approved, 'Rating) > > > > > > val resultTable = tableEnv.sql( > > > "SELECT ProductID, AVG(Rating) FROM Reviews WHERE Approved = true > > > GROUP BY ProductID" > > > ) > > > > > > val typeInfo = createTypeInformation[(ProductId, Double)] > > > val outStream = resultTable.toRetractStream(typeInfo) > > > > > > outStream.print() > > > > > > env.execute("Flink SQL Average rating") > > > > > > } > > > } > > > > > > > > > > > > > > > > > > > > > > > >
