You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2020/12/08 15:14:04 UTC
[GitHub] [flink] sjwiesman commented on a change in pull request #14292: [FLINK-20456][docs] Make streaming SQL concepts more approachable
sjwiesman commented on a change in pull request #14292:
URL: https://github.com/apache/flink/pull/14292#discussion_r538482958
##########
File path: docs/dev/table/streaming/temporal_tables.md
##########
@@ -57,134 +56,68 @@ SELECT * FROM product_changelog;
-(DELETE) 18:00:00 p_001 scooter 12.99
{% endhighlight %}
-The table `product_changelog` represents an ever growing changelog of database table `products`, for example, the initial price of product `scooter` is `11.11` at `00:01:00`, and the price increases to `12.99` at `12:00:00`,
- the product item is deleted from the table `products` at `18:00:00`.
+Given this set of changes, we track how the price of a scooter changes over time.
+It is initially $11.11 at `00:01:00` when added to the catalog.
+The price then rises to $12.99 at `12:00:00` before being deleted from the catalog at `18:00:00`.
-Given that we would like to output the version of `product_changelog` table of the time `10:00:00`, the following table shows the result.
-{% highlight sql %}
-update_time product_id product_name price
-=========== ========== ============ =====
-00:01:00 p_001 scooter 11.11
-00:02:00 p_002 basketball 23.11
-{% endhighlight %}
+If we queried the table for various products' prices at different times, we would retrieve different results. At `10:00:00` the table would show one set of prices.
-Given that we would like to output the version of `product_changelog` table of the time `13:00:00`, the following table shows the result.
{% highlight sql %}
update_time product_id product_name price
=========== ========== ============ =====
12:00:00 p_001 scooter 12.99
12:00:00 p_002 basketball 19.99
{% endhighlight %}
-In above example, the specific version of the table is tracked by `update_time` and `product_id`, the `product_id` would be a primary key for `product_changelog` table and `update_time` would be the event time.
-
-In Flink, this is represented by a [*versioned table*](#defining-versioned-table).
-
-### Correlate with a regular table
-On the other hand, some use cases require to join a regular table which is an external database table.
+While at `13:00:00,` we found find another.
-Let's assume that `LatestRates` is a table (e.g. stored in HBase) which is materialized with the latest rates. The `LatestRates` always represents the latest content of hbase table `rates`.
-
-Then the content of `LatestRates` table when we query at time `10:15:00` is:
{% highlight sql %}
-10:15:00 > SELECT * FROM LatestRates;
-
-currency rate
-========= ====
-US Dollar 102
-Euro 114
-Yen 1
-{% endhighlight %}
-
-Then the content of `LatestRates` table when we query at time `11:00:00` is:
-{% highlight sql %}
-11:00:00 > SELECT * FROM LatestRates;
-
-currency rate
-========= ====
-US Dollar 102
-Euro 116
-Yen 1
+update_time product_id product_name price
+=========== ========== ============ =====
+12:00:00 p_001 scooter 12.99
+12:00:00 p_002 basketball 19.99
{% endhighlight %}
-In Flink, this is represented by a [*regular Table*](#defining-regular-table).
-Temporal Table
---------------
-<span class="label label-danger">Attention</span> This is only supported in Blink planner.
+## Versioned Table Sources
-Flink uses primary key constraint and event time to define both versioned table and versioned view.
+Versioned tables are defined implicity for any table whose underlying source or format directly define changelogs. Examples include the [upsert Kafka]({% link dev/table/connectors/upsert-kafka.md %}) source as well as database changelog formats such as [debezium]({% link dev/table/connectors/formats/debezium.md %}) and [canal]({% link dev/table/connectors/formats/canal.md %}). As discussed above, the only additional requirement is the `CREATE` table statement must contain a primary key and event-time attribute.
-### Defining Versioned Table
-The table is a versioned table in Flink only is the table contains primary key constraint and event time.
{% highlight sql %}
--- Define a versioned table
-CREATE TABLE product_changelog (
- product_id STRING,
- product_name STRING,
- product_price DECIMAL(10, 4),
- update_time TIMESTAMP(3) METADATA FROM 'value.source.timestamp' VIRTUAL,
- PRIMARY KEY(product_id) NOT ENFORCED, -- (1) defines the primary key constraint
- WATERMARK FOR update_time AS update_time -- (2) defines the event time by watermark
-) WITH (
- 'connector' = 'kafka',
- 'topic' = 'products',
- 'scan.startup.mode' = 'earliest-offset',
- 'properties.bootstrap.servers' = 'localhost:9092',
- 'value.format' = 'debezium-json'
-);
+CREATE TABLE products (
+ product_id STRING,
+ product_name STRING,
+ price DECIMAL(32, 2),
+ update_time TIMESTAMP(3) METADATA FROM 'value.source.timestamp' VIRTUAL,
+ PRIMARY KEY (product_id) NOT ENFORCED
+ WATERMARK FOR update_time AS update_time
+) WITH (...);
{% endhighlight %}
-Line `(1)` defines the primary key constraint for table `product_changelog`, Line `(2)` defines the `update_time` as event time for table `product_changelog`,
-thus table `product_changelog` is a versioned table.
+## Versioned Table Views
-**Note**: The grammar `METADATA FROM 'value.source.timestamp' VIRTUAL` means extract the database
-operation execution time for every changelog, it's strongly recommended defines the database operation execution time
-as event time rather than ingestion-time or time in the record, otherwise the version extract from the changelog may
-mismatch with the version in database.
-
-### Defining Versioned View
+Flink also supports defining versioned views if the underlying query contains a unique key constraint and event-time attribute. Image an append-only table of currency rates.
-Flink also supports defining versioned view only if the view contains unique key constraint and event time.
-
-Let’s assume that we have the following table `RatesHistory`:
{% highlight sql %}
--- Define an append-only table
-CREATE TABLE RatesHistory (
- currency_time TIMESTAMP(3),
- currency STRING,
- rate DECIMAL(38, 10),
- WATERMARK FOR currency_time AS currency_time -- defines the event time
+CREATE TABLE currency_rates (
+ currency STRING,
+ rate DECIMAL(32, 10)
+ update_time TIMESTAMP(3),
+ WATERMARK FOR update_time AS update_time
) WITH (
- 'connector' = 'kafka',
- 'topic' = 'rates',
- 'scan.startup.mode' = 'earliest-offset',
- 'properties.bootstrap.servers' = 'localhost:9092',
- 'format' = 'json' -- this is an append only source
-)
+ 'connector' = 'kafka',
+ 'topic' = 'rates',
+ 'properties.bootstrap.servers' = 'localhost:9092',
+ 'format' = 'json'
+);
{% endhighlight %}
-Table `RatesHistory` represents an ever growing append-only table of currency exchange rates with respect to
-Yen (which has a rate of 1). For example, the exchange rate for the period from 09:00 to 10:45 of Euro to Yen was 114.
-From 10:45 to 11:15 it was 116.
-
-{% highlight sql %}
-SELECT * FROM RatesHistory;
-
-currency_time currency rate
-============= ========= ====
-09:00:00 US Dollar 102
-09:00:00 Euro 114
-09:00:00 Yen 1
-10:45:00 Euro 116
-11:15:00 Euro 119
-11:49:00 Pounds 108
-{% endhighlight %}
+The table `currency_rates` contains a row for each currency - with respect to USD - and receives a new row each time the rate changes.
+The `JSON` format does not support native changelog semantics, and so Flink can only read this table as append-only.
+However, it is clear to us (the query developer) that this table has all the necessary information to define a versioned table.
-To define a versioned table on `RatesHistory`, Flink supports defining a versioned view
-by [deduplication query]({% link dev/table/sql/queries.md %}#deduplication) which produces an ordered changelog
-stream with an inferred primary key(`currency`) and event time(`currency_time`).
+Flink can reinterpret this table as a versioned table by defining a [deduplication query]({% link dev/table/sql/queries.md %}#deduplication) which produces an ordered changelog stream with an inferred primary key (currency) and event time (update_time).
Review comment:
The real answer is flink has a special optimization step that turns any `OVER WINDOW` with exactly this format into a versioned table keyed by the fields in the `PARTITION BY` clause.
```sql
SELECT *,
ROW_NUMBER() OVER (PARTITION BY [key,] ORDER BY [time attribute] DESC) AS row_num
WHERE
row_num = 1
```
In a batch query, this would return the most recent value for each currency. The streaming semantic is it provides the new value for each partition each time a more recent value is made available.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org