From 6b468b939864c496822aa858f4aa21c7ff39505a Mon Sep 17 00:00:00 2001 From: discord9 Date: Fri, 28 Aug 2026 14:31:18 +0800 Subject: [PATCH] docs: update Flow user guide Signed-off-by: discord9 --- .../continuous-aggregation.md | 42 ++--- .../flow-computation/expressions.md | 10 +- .../flow-computation/manage-flow.md | 158 +++++++++--------- docs/user-guide/flow-computation/overview.md | 8 +- .../continuous-aggregation.md | 29 ++-- .../flow-computation/expressions.md | 14 +- .../flow-computation/manage-flow.md | 128 +++++++------- .../user-guide/flow-computation/overview.md | 8 +- 8 files changed, 201 insertions(+), 196 deletions(-) diff --git a/docs/user-guide/flow-computation/continuous-aggregation.md b/docs/user-guide/flow-computation/continuous-aggregation.md index 64e0c7dc6d..7e4644ac73 100644 --- a/docs/user-guide/flow-computation/continuous-aggregation.md +++ b/docs/user-guide/flow-computation/continuous-aggregation.md @@ -59,7 +59,7 @@ CREATE TABLE `ngx_statistics` ( Then create the flow `ngx_aggregation` to aggregate a series of aggregate functions, including `count`, `min`, `max`, `avg` of the `size` column, and the number of packets with a size greater than 550. The aggregation is calculated in 1-minute fixed windows of `access_time` column and also grouped by the `status` column. A spike in `high_size_count` or `max_size` within a single window then points you at the minute to inspect. -The `EXPIRE AFTER '6h'` in the following SQL ensures that the flow computation only uses source data from the last 6 hours. Data older than 6 hours in the sink table will not be modified by this flow. For more details, see [manage-flow](manage-flow.md#expire-after). +The `EXPIRE AFTER '6h'` in the following SQL ensures that the flow computation only uses source data from the last 6 hours. Data older than 6 hours in the sink table will not be modified by this flow. For more details, see [Manage Flow](manage-flow.md#expire-after). ```sql CREATE FLOW ngx_aggregation @@ -74,7 +74,7 @@ SELECT max(size) as max_size, avg(size) as avg_size, sum(case when `size` > 550 then 1 else 0 end) as high_size_count, - date_bin('1 minutes'::INTERVAL, access_time) as time_window, + date_bin('1 minutes'::INTERVAL, access_time) as time_window FROM ngx_access_log GROUP BY status, @@ -92,7 +92,9 @@ VALUES ('ios', 'iOS', 'referer', 'GET', '/api/v1', 'trace_id', 'HTTP', 404, 700, 'agent', now()); ``` -Then the sink table `ngx_statistics` will be incremental updated and contain the following data: +Then the sink table `ngx_statistics` will be incrementally updated and contain the following data: + +The time-window and `update_at` timestamps in these result tables are illustrative and vary with execution time. ```sql SELECT * FROM ngx_statistics; @@ -118,7 +120,7 @@ VALUES ('ios', 'iOS', 'referer', 'GET', '/api/v1', 'trace_id', 'HTTP', 404, 800, 'agent', now()); ``` -The sink table `ngx_statistics` now have corresponding rows updated, notes how `max_size`, `avg_size` and `high_size_count` are updated: +The sink table `ngx_statistics` now has corresponding rows updated; note how `max_size`, `avg_size` and `high_size_count` are updated: ```sql SELECT * FROM ngx_statistics; @@ -152,7 +154,7 @@ Another example of real-time analytics is to retrieve all distinct countries fro You can use the following query to group countries by time window: ```sql -/* input table */ +/* source table */ CREATE TABLE ngx_access_log ( client STRING, country STRING, @@ -179,7 +181,7 @@ COMMENT 'aggregate for distinct country' AS SELECT DISTINCT country, - date_bin('1 hour'::INTERVAL, access_time) as time_window, + date_bin('1 hour'::INTERVAL, access_time) as time_window FROM ngx_access_log GROUP BY country, @@ -190,7 +192,7 @@ The above query puts the data from the `ngx_access_log` table into the `ngx_coun It calculates the distinct country for each time window. The `date_bin` function is used to group the data into one-hour intervals. The `ngx_country` table will be continuously updated with the aggregated data, -providing real-time insights into the distinct countries that are accessing the system. The `EXPIRE AFTER` make flow ignore data with `access_time` older than 7 days and no longer calculate them anymore, see more explain in [manage-flow](manage-flow.md#expire-after). +providing real-time insights into the distinct countries that are accessing the system. The `EXPIRE AFTER` clause makes the Flow ignore data with `access_time` older than 7 days; see [Manage Flow](manage-flow.md#expire-after) for details. You can insert some data into the source table `ngx_access_log`: @@ -233,7 +235,7 @@ select * from ngx_country; Consider a usecase where you have a stream of sensor events from a network of temperature sensors that you want to monitor in real-time. The sensor events contain information such as the sensor ID, the temperature reading, the timestamp of the reading, and the location of the sensor. You want to continuously aggregate this data to provide real-time alerts when the temperature exceeds a certain threshold. Then the query for continuous aggregation would be: ```sql -/* create input table */ +/* create source table */ CREATE TABLE temp_sensor_data ( sensor_id INT, loc STRING, @@ -263,7 +265,7 @@ SELECT sensor_id, loc, max(temperature) as max_temp, - date_bin('10 seconds'::INTERVAL, ts) as time_window, + date_bin('10 seconds'::INTERVAL, ts) as time_window FROM temp_sensor_data GROUP BY sensor_id, @@ -276,7 +278,7 @@ The above query continuously aggregates data from the `temp_sensor_data` table i It calculates the maximum temperature reading for each sensor and location, filtering out data where the maximum temperature exceeds 100 degrees. The `temp_alerts` table will be continuously updated with the aggregated data, -providing real-time alerts (in the form of new rows in the `temp_alerts` table) when the temperature exceeds the threshold. The `EXPIRE AFTER '1h'` makes flow only calculate source data with `ts` in `(now - 1h, now)` range, see more explain in [manage-flow](manage-flow.md#expire-after). +providing real-time alerts (in the form of new rows in the `temp_alerts` table) when the temperature exceeds the threshold. The `EXPIRE AFTER '1h'` makes flow only calculate source data with `ts` in `(now - 1h, now)` range, see [Manage Flow](manage-flow.md#expire-after) for details. Now that we have created the flow task, we can insert some data into the source table `temp_sensor_data`: @@ -286,7 +288,7 @@ INSERT INTO temp_sensor_data VALUES (1, 'room1', 98.5, now() - '10 second'::INTERVAL), (2, 'room2', 99.5, now()); ``` -table should be empty now, but still wait at least few seconds for flow to update results to sink table: +The table should be empty now; wait a few seconds for the Flow to update the sink table: ```sql SELECT * FROM temp_alerts; @@ -304,7 +306,7 @@ INSERT INTO temp_sensor_data VALUES (2, 'room2', 102.5, now()); ``` -wait at least few seconds for flow to update results to sink table: +Wait a few seconds for the Flow to update the sink table: ```sql SELECT * FROM temp_alerts; @@ -325,7 +327,7 @@ SELECT * FROM temp_alerts; Consider a usecase in which you need a bar graph that show the distribution of packet sizes for each status code to monitor the health of the system. The query for continuous aggregation would be: ```sql -/* create input table */ +/* create source table */ CREATE TABLE ngx_access_log ( client STRING, stat INT, @@ -352,7 +354,7 @@ SELECT stat, trunc(size, -1)::INT as bucket_size, count(client) AS total_logs, - date_bin('1 minutes'::INTERVAL, access_time) as time_window, + date_bin('1 minutes'::INTERVAL, access_time) as time_window FROM ngx_access_log GROUP BY @@ -364,7 +366,7 @@ GROUP BY The query aggregates data from the `ngx_access_log` table into the `ngx_distribution` table. It computes the total number of logs for each status code and packet size bucket (bucket size of 10, as specified by `trunc` with a second argument of -1) within each time window. The `date_bin` function groups the data into one-minute intervals. -The `EXPIRE AFTER '6h'` ensures that the flow computation only uses source data from the last 6 hours. See more details in [manage-flow](manage-flow.md#expire-after). +The `EXPIRE AFTER '6h'` ensures that the flow computation only uses source data from the last 6 hours. See more details in [Manage Flow](manage-flow.md#expire-after). Now that we have created the flow task, we can insert some data into the source table `ngx_access_log`: @@ -381,7 +383,7 @@ INSERT INTO ngx_access_log VALUES ('cli9', 404, 180, now()), ('cli10', 404, 184, now()); ``` -wait at least few seconds for flow to update results to sink table: +Wait a few seconds for the Flow to update the sink table: ```sql SELECT * FROM ngx_distribution; @@ -441,8 +443,8 @@ This table will serve as the data source for our TQL-based Flow computations. Th Now we'll create a Flow that uses TQL to calculate the rate of `byte` over time: ```sql -CREATE FLOW calc_rate -SINK TO rate_reqs +CREATE FLOW calc_rate +SINK TO rate_reqs EVAL INTERVAL '1m' AS TQL EVAL (now() - '1m'::interval, now(), '30s') rate(http_requests_total{job="my_service"}[1m]); ``` @@ -574,8 +576,10 @@ When you're done experimenting, clean up the resources: ```sql DROP FLOW calc_rate; -DROP TABLE http_requests; +DROP FLOW calc_rate_cte; +DROP TABLE http_requests_total; DROP TABLE rate_reqs; +DROP TABLE rate_reqs_cte; ``` ## Next Steps diff --git a/docs/user-guide/flow-computation/expressions.md b/docs/user-guide/flow-computation/expressions.md index c34230caa9..eaa379a078 100644 --- a/docs/user-guide/flow-computation/expressions.md +++ b/docs/user-guide/flow-computation/expressions.md @@ -7,15 +7,15 @@ description: Lists supported aggregate and scalar functions in GreptimeDB's flow ## Aggregate functions -Flow support all aggregate functions that a normal sql query supports such as `COUNT`, `SUM`, `MIN`, `MAX`, etc. For a detailed list, please refer to [Aggregate Functions](/reference/sql/functions/df-functions.md#aggregate-functions). +Flow supports aggregate functions that are supported by the SQL query engine and the Flow plan, such as `COUNT`, `SUM`, `MIN`, and `MAX`. Unsupported query plans fail when the Flow is created. For a detailed list, please refer to [Aggregate Functions](/reference/sql/functions/df-functions.md#aggregate-functions). ## Scalar functions -Flow support all scalar functions that a normal sql query supports in our [SQL reference](/reference/sql/functions/overview.md). +Flow supports scalar functions that are supported by the SQL query engine and the Flow plan. Unsupported query plans fail when the Flow is created. See our [SQL reference](/reference/sql/functions/overview.md) for the function catalogue. -And here are some of the most commonly used scalar functions in flow: +Here are some commonly used scalar functions in Flow: -- [`date_bin`](/reference/sql/functions/df-functions.md#date_bin): calculate time intervals and returns the start of the interval nearest to the specified timestamp. +- [`date_bin`](/reference/sql/functions/df-functions.md#date_bin): calculates time intervals and returns the start of the interval nearest to the specified timestamp. - [`date_trunc`](/reference/sql/functions/df-functions.md#date_trunc): truncate a timestamp value to a specified precision. -- [`trunc`](/reference/sql/functions/df-functions.md#trunc): truncate a number to a whole number or truncated to the specified decimal places. +- [`trunc`](/reference/sql/functions/df-functions.md#trunc): truncate a number to a whole number or to the specified decimal places. diff --git a/docs/user-guide/flow-computation/manage-flow.md b/docs/user-guide/flow-computation/manage-flow.md index 3b0f42d5c0..6e57f06c76 100644 --- a/docs/user-guide/flow-computation/manage-flow.md +++ b/docs/user-guide/flow-computation/manage-flow.md @@ -26,32 +26,21 @@ CREATE TABLE temp_sensor_data ( PRIMARY KEY(sensor_id, loc) ); ``` -Avoid using `WITH ('ttl' = 'instant')` for new Flow source tables. Source tables with `ttl='instant'` fall back to the deprecated streaming mode. Keep the source data with an appropriate TTL instead, so aggregation and TQL Flow workloads can run in batching mode. - -For existing legacy streaming-mode deployments, a source table may use `WITH ('ttl' = 'instant')`: - -```sql -CREATE TABLE temp_sensor_data ( - sensor_id INT, - loc STRING, - temperature DOUBLE, - ts TIMESTAMP TIME INDEX, - PRIMARY KEY(sensor_id, loc) -) WITH ('ttl' = 'instant'); -``` - -Setting `'ttl'` to `'instant'` makes the table discard inserted data immediately and only sends rows to a legacy streaming-mode flow task. This pattern is deprecated for new workloads. +For new workloads, avoid `WITH ('ttl' = 'instant')` on Flow source tables. This is a legacy pattern and is not recommended for new aggregation or TQL workloads. Keep source data with an appropriate retention policy instead. ## Create a Sink Table -A flow stores its aggregated data in a sink table. `CREATE FLOW` creates that table when it does not exist, -so defining it yourself is optional — do it when you want control over the schema, for example to choose the -primary key or set a TTL. Either way, the sink table has to be compatible with the flow's query result: +A flow stores its aggregated data in a sink table. When the sink table does not exist, `CREATE FLOW` +automatically creates it when the query result is sufficient to infer its schema. Pre-create the sink when you +need control over its schema or layout, or when inference is complex. An existing sink table is validated against +the flow's query result. The source and sink tables must be different tables. + +The sink table has to be compatible with the flow's query result: -- **Column order and type**: Ensure the order and type of the columns in the sink table match the query result of the flow. +- **Column order and type**: For a pre-created SQL sink, match the query output columns in order and type. - **Time index**: Specify the `TIME INDEX` for the sink table, typically using the time window column generated by the time window function. -- **Update time**: The Flow engine automatically appends the update time to the end of each computation result row. This update time is stored in the `update_at` column. Ensure that this column is included in the sink table schema. -- **Tags**: Use `PRIMARY KEY` to specify Tags, which together with the time index serves as a unique identifier for row data and optimizes query performance. +- **Update time**: For an auto-created batching SQL sink, Flow adds an `update_at` column for the update time. TQL sinks follow the query output and do not automatically add `update_at`. A pre-created SQL sink can either match the query output width or include one extra trailing timestamp column for update time. +- **Tags**: Use `PRIMARY KEY` to specify Tags, which together with the time index serve as a unique identifier for row data and optimize query performance. For example: @@ -94,43 +83,39 @@ The sink table has the columns `sensor_id`, `loc`, `max_temp`, `time_window`, an The grammar to create a flow is: - - ```sql CREATE [ OR REPLACE ] FLOW [ IF NOT EXISTS ] SINK TO [ EXPIRE AFTER ] +[ EVAL INTERVAL ] [ COMMENT '' ] [ WITH ( = [, ...]) ] -AS +AS ; ``` +The clauses must appear in the order shown: `EXPIRE AFTER` comes before `EVAL INTERVAL`. +`EVAL INTERVAL` schedules repeated evaluation of the full query. Scheduled SQL flows can use joins, +subqueries, and SQL CTEs when the SQL query engine can plan the query. TQL flows require `EVAL INTERVAL`. +Batching time-window aggregate flows can run without `EVAL INTERVAL`. + When `OR REPLACE` is specified, any existing flow with the same name will be updated to the new version. It's important to note that this only affects the flow task itself; the source and sink tables will remain unchanged. Conversely, when `IF NOT EXISTS` is specified, the command will have no effect if the flow already exists, rather than reporting an error. Additionally, please note that `OR REPLACE` cannot be used in conjunction with `IF NOT EXISTS`. -- `flow-name` is an unique identifier in the catalog level. +- `flow-name` is a unique identifier at the catalog level. - `sink-table-name` is the table name where the materialized aggregated data is stored. - It can be an existing table or a new one. `flow` will create the sink table if it doesn't exist. - -- `EXPIRE AFTER` is an optional interval to expire the data from the Flow engine. - For more details, please refer to the [`EXPIRE AFTER`](#expire-after) part. + It can be an existing table or a new one; see [Create a Sink Table](#create-a-sink-table) for creation and validation behavior. +- `EXPIRE AFTER` is an optional interval to expire data from the Flow engine. For more details, please refer to the [`EXPIRE AFTER`](#expire-after) section. +- `EVAL INTERVAL` is an optional interval for scheduled full-query evaluation. TQL flows require it. - `COMMENT` is the description of the flow. - `WITH` specifies flow options. - For example, the experimental `experimental_enable_incremental_read` option enables incremental source reads for eligible batching flows. + The user-facing options documented below are `defer_on_missing_source` and the experimental `experimental_enable_incremental_read`. - `SQL` part defines the continuous aggregation query. - It defines the source tables provide data for the flow. + It defines the source tables that provide data for the flow. Each flow can have multiple source tables. - Please Refer to [Write a Query](#write-a-sql-query) for the details. + Please refer to [Write a SQL query](#write-a-sql-query) for details. A simple example to create a flow: @@ -142,31 +127,40 @@ COMMENT 'My first flow in GreptimeDB' AS SELECT max(temperature) as max_temp, - date_bin('10 seconds'::INTERVAL, ts) as time_window, + date_bin('10 seconds'::INTERVAL, ts) as time_window FROM temp_sensor_data GROUP BY time_window; ``` -The created flow will compute `max(temperature)` for every 10 seconds and store the result in `my_sink_table`. All data comes within 1 hour will be used in the flow. +The created flow groups `max(temperature)` into 10-second windows and stores the result in `my_sink_table`. Data within the last hour is used in the flow. ### EXPIRE AFTER -The `EXPIRE AFTER` clause specifies the interval after which data will expire from the flow engine. +The `EXPIRE AFTER` clause specifies the interval after which data will expire from the flow engine. -Data in the source table that exceeds the specified expiration time will no longer be included in the flow's calculations. -Similarly, data in the sink table that is older than the expiration time will not be updated. -This means the flow engine will ignore data older than the specified interval during aggregation. -This mechanism helps to manage the state size for stateful queries, such as those involving `GROUP BY`. +For a Flow with a usable time-window expression, data in the source table older than the specified interval is excluded from calculations, and older sink rows are not updated. This limits the state and recomputation range for time-window flows, including stateful queries such as those involving `GROUP BY`. -It is important to note that the `EXPIRE AFTER` clause does not delete data from either the source table or the sink table. -It only controls how the flow engine processes the data. -If you want to delete data from the source or sink table, please [set the `TTL` option](/user-guide/manage-data/overview.md#manage-data-retention-with-ttl-policies) when creating tables. +Scheduled full-query SQL and TQL flows execute unfiltered snapshots unless the query contains its own time predicate; `EXPIRE AFTER` does not add a time filter. It does not delete data from either table. If you want to delete data from the source or sink table, please [set the `TTL` option](/user-guide/manage-data/overview.md#manage-data-retention-with-ttl-policies) when creating tables. Setting a reasonable time interval for `EXPIRE AFTER` is helpful to limit how far back the batching engine needs to recompute results and to avoid excessive resource usage. It serves a similar purpose to bounding lateness in stream processing systems, but new Flow workloads should use batching mode. For example, if the flow engine processes the aggregation at 10:00:00 and the `'1 hour'::INTERVAL` is set, -any input data that arrive now with a time index older than 1 hour (before 09:00:00) will expire and be ignore. -Only data timestamped from 09:00:00 onwards will be used in the aggregation and update to sink table. +any input data that arrive now with a time index older than 1 hour (before 09:00:00) will expire and be ignored. +Only data timestamped from 09:00:00 onwards will be used in the aggregation and to update the sink table. + +### Defer creation when a source is missing + +By default, creating a Flow fails if one of its source tables does not exist. Set +`defer_on_missing_source` to `true` to persist a pending Flow instead of failing. The Flow is not scheduled while its +sources remain unresolved. + +```sql +CREATE FLOW pending_flow +SINK TO pending_sink +WITH (defer_on_missing_source = 'true') +AS +SELECT * FROM source_created_later; +``` ### Experimental incremental source reads @@ -202,47 +196,32 @@ GROUP BY time_window; ``` -When this option is enabled, Flow keeps per-region source sequence watermarks and attempts to read only newly appended source rows after the initial full snapshot. -This is an execution optimization and does not change the query result. +When enabled, Flow attempts to read only newly appended source rows after the initial full snapshot. +This is an execution optimization and does not change the query result. The optimization is not a persistence +contract: the first run, and a run after a restart or when incremental reading is not safe, may use a full snapshot. The current limitations are: - All source tables must be append-only tables created with `append_mode = 'true'`. Flow creation fails if any source table is not append-only. -- The optimization only applies to batching SQL flows. - TQL flows, unsupported aggregate shapes, and simple projection/filter flows do not use incremental source reads. -- Source tables created with `ttl = 'instant'` currently use streaming mode and do not use this batching-mode option. -- The first run still needs a full snapshot. - Later runs may fall back to full snapshot or retry/repair when GreptimeDB cannot safely use incremental source reads. +- The optimization applies only to eligible batching SQL flows. TQL flows and plans that do not support incremental + reads use the normal full-snapshot behavior. ### Write a SQL query -The `SQL` part of the flow is similar to a standard `SELECT` clause with a few differences. The syntax of the query is as follows: +The SQL after `AS` is planned as a standard SQL query. A typical batching time-window aggregate has this shape: ```sql -SELECT AGGR_FUNCTION(column1, column2,..) [, TIME_WINDOW_FUNCTION() as time_window] FROM GROUP BY {time_window | column1, column2,.. }; +SELECT AGGR_FUNCTION(column1, column2,..) [, TIME_WINDOW_FUNCTION() as time_window] +FROM +GROUP BY {time_window | column1, column2,.. }; ``` -Only the following types of expressions are allowed after the `SELECT` keyword: -- Aggregate functions: Refer to the [Expressions](expressions.md) documentation for details. -- Time window functions: Refer to the [define time window](#define-time-window) section for details. -- Scalar functions: Such as `col`, `to_lowercase(col)`, `col + 1`, etc. This part is the same as in a standard `SELECT` clause in GreptimeDB. - -The following points should be noted about the rest of the query syntax: -- The query must include a `FROM` clause to specify the source table. - As join clauses are currently not supported, - the query can only aggregate columns from a single table. -- `WHERE` and `HAVING` clauses are supported. - The `WHERE` clause filters data before aggregation, - while the `HAVING` clause filters data after aggregation. -- `DISTINCT` currently only works with the `SELECT DISTINCT column1 ..` syntax. - It is used to remove duplicate rows from the result set. - Support for `SELECT count(DISTINCT column1) ...` is not available yet but will be added in the future. -- The `GROUP BY` clause works the same as a standard queries, - grouping data by specified columns. - The time window column in the `GROUP BY` clause is crucial for continuous aggregation scenarios. - Other expressions in `GROUP BY` can include literals, columns, or scalar expressions. -- `ORDER BY`, `LIMIT`, and `OFFSET` are not supported. +The query engine and Flow plan determine which SQL expressions and clauses are supported. For a scheduled full-query +SQL Flow, planner-valid joins, subqueries, and SQL CTEs are supported; an unsupported plan fails when the Flow is +created. For batching time-window aggregates, `GROUP BY` commonly includes the time-window expression. See +[Expressions](expressions.md) for functions commonly used in Flow queries, and [Define time window](#define-time-window) +for fixed windows. Refer to [Continuous Aggregation](continuous-aggregation.md) for more examples of how to use continuous aggregation in real-time analytics, monitoring, and dashboards. @@ -275,9 +254,26 @@ For more details on the behavior of the function, please refer to [`date_bin`](/reference/sql/functions/df-functions.md#date_bin). :::tip NOTE -Currently, flow rely on the time window expr to determine how to incrementally update the result. So it's better to use a relatively small time window when possible. +The time-window expression helps Flow determine how to update results incrementally. The appropriate window size +depends on the workload and query semantics. ::: +## Inspect flows + +Use the following statements and system tables to inspect Flow definitions and runtime information: + +```sql +SHOW FLOWS; +SHOW CREATE FLOW my_flow; +SHOW FLOW STATUS LIKE 'my%'; +SELECT * FROM information_schema.flows; +SELECT * FROM information_schema.flow_statistics; +``` + +`SHOW FLOWS` lists flows, `SHOW CREATE FLOW` returns a Flow definition, and `SHOW FLOW STATUS` returns runtime +statistics. The `information_schema` tables provide definition and statistics details. Runtime fields can initially be +`NULL`, and values can lag behind the latest state in distributed deployments. + ## Flush a flow The flow engine automatically processes aggregation operations within a short period(i.e. few seconds) when new data arrives in the source table. diff --git a/docs/user-guide/flow-computation/overview.md b/docs/user-guide/flow-computation/overview.md index e77ef02bc1..88201a2d99 100644 --- a/docs/user-guide/flow-computation/overview.md +++ b/docs/user-guide/flow-computation/overview.md @@ -7,7 +7,7 @@ description: Discover how GreptimeDB's Flow engine enables real-time continuous GreptimeDB's Flow engine enables real-time computation on incoming data. It is particularly beneficial for Extract-Transform-Load (ETL) processes or for performing continuous aggregations such as sum, average, and other time-window calculations. -Each new row updates the sink table incrementally, so the aggregation is computed on write rather than on every query. +Flow materializes computation results in the sink table as source data is processed, so queries can read the computed results instead of recalculating them from raw data. Use cases include: @@ -20,10 +20,8 @@ Use cases include: Flow uses batching mode for aggregation and TQL workloads. Simple non-aggregation Flow queries currently use the deprecated streaming mode and are not recommended for new workloads. ::: -Upon data insertion into the source table, -the data is concurrently ingested to the Flow engine. -At each trigger interval (one second), -the Flow engine executes the specified computations and updates the sink table with the results. +Upon data insertion into the source table, the data is made available to the Flow engine. +Flow then processes the specified computation and updates the sink table with the results. Both the source and sink tables are time-series tables within GreptimeDB. Before creating a Flow, it is crucial to define the schemas for these tables and design the Flow to specify the computation logic. diff --git a/i18n/zh/docusaurus-plugin-content-docs/current/user-guide/flow-computation/continuous-aggregation.md b/i18n/zh/docusaurus-plugin-content-docs/current/user-guide/flow-computation/continuous-aggregation.md index f01c212d62..e9dc930da9 100644 --- a/i18n/zh/docusaurus-plugin-content-docs/current/user-guide/flow-computation/continuous-aggregation.md +++ b/i18n/zh/docusaurus-plugin-content-docs/current/user-guide/flow-computation/continuous-aggregation.md @@ -17,7 +17,7 @@ Flow 引擎按时间窗口维护总和、平均值、计数等聚合结果，新 ### 日志统计 -这个例子是根据输入表中的数据计算一系列统计数据，包括一分钟时间窗口内的总日志数、最小大小、最大大小、平均大小以及大小大于 550 的数据包数。 +这个例子是根据 source 表中的数据计算一系列统计数据，包括一分钟时间窗口内的总日志数、最小大小、最大大小、平均大小以及大小大于 550 的数据包数。 首先，创建一个 source 表 `ngx_access_log` 和一个 sink 表 `ngx_statistics`，如下所示： @@ -72,7 +72,7 @@ SELECT max(size) as max_size, avg(size) as avg_size, sum(case when `size` > 550 then 1 else 0 end) as high_size_count, - date_bin('1 minutes'::INTERVAL, access_time) as time_window, + date_bin('1 minutes'::INTERVAL, access_time) as time_window FROM ngx_access_log GROUP BY status, @@ -90,7 +90,9 @@ VALUES ('ios', 'iOS', 'referer', 'GET', '/api/v1', 'trace_id', 'HTTP', 404, 700, 'agent', now()); ``` -则 `ngx_access_log` 表将被增量更新以包含以下数据： +然后，sink 表 `ngx_statistics` 将被增量更新并包含以下数据。 + +下面结果中的时间窗口和 `update_at` 时间戳仅用于示例，会随执行时间而变化。 ```sql SELECT * FROM ngx_statistics; @@ -176,7 +178,7 @@ COMMENT 'aggregate for distinct country' AS SELECT DISTINCT country, - date_bin('1 hour'::INTERVAL, access_time) as time_window, + date_bin('1 hour'::INTERVAL, access_time) as time_window FROM ngx_access_log GROUP BY country, @@ -185,7 +187,7 @@ GROUP BY 上述查询将 `ngx_access_log` 表中的数据聚合到 `ngx_country` 表中，它计算了每个时间窗口内的不同国家。 `date_bin` 函数用于将数据聚合到一小时的间隔中。 -`ngx_country` 表将不断更新聚合数据，以监控访问系统的不同国家。`EXPIRE AFTER` 参数将确保流式处理流程自动忽略 `access_time` 超过 7 天的数据且不再参与 flow 计算，详见 请参阅[管理 Flow](manage-flow.md#expire-after) 中的说明。 +`ngx_country` 表将不断更新聚合数据，以监控访问系统的不同国家。`EXPIRE AFTER` 参数会使 Flow 忽略 `access_time` 早于 7 天的数据，详见[管理 Flow](manage-flow.md#expire-after)。 你可以向 source 表 `ngx_access_log` 插入一些数据： @@ -257,7 +259,7 @@ SELECT sensor_id, loc, max(temperature) as max_temp, - date_bin('10 seconds'::INTERVAL, ts) as time_window, + date_bin('10 seconds'::INTERVAL, ts) as time_window FROM temp_sensor_data GROUP BY sensor_id, @@ -279,7 +281,7 @@ INSERT INTO temp_sensor_data VALUES (2, 'room2', 99.5, now()); ``` -表现在应该是空的，等待几秒钟让 flow 将结果更新到输出表： +此时表应为空；等待几秒钟让 Flow 更新 sink 表： ```sql SELECT * FROM temp_alerts; @@ -297,7 +299,7 @@ INSERT INTO temp_sensor_data VALUES (2, 'room2', 102.5, now()); ``` -等待几秒钟，让 flow 将结果更新到输出表： +等待几秒钟，让 Flow 更新 sink 表： ```sql SELECT * FROM temp_alerts; @@ -343,7 +345,7 @@ SELECT stat, trunc(size, -1)::INT as bucket_size, count(client) AS total_logs, - date_bin('1 minutes'::INTERVAL, access_time) as time_window, + date_bin('1 minutes'::INTERVAL, access_time) as time_window FROM ngx_access_log GROUP BY @@ -373,7 +375,7 @@ INSERT INTO ngx_access_log VALUES ('cli10', 404, 184, now()); ``` -等待几秒钟，让 flow 将结果更新到 sink 表： +等待几秒钟，让 Flow 更新 sink 表： ```sql SELECT * FROM ngx_distribution; @@ -394,7 +396,6 @@ SELECT * FROM ngx_distribution; ``` - ## 将 TQL 与 Flow 结合使用进行高级时序分析 :::warning 实验性特性 @@ -413,7 +414,7 @@ TQL 与 Flow 的集成提供了以下几个优势： 3. **连续处理**：结合 Flow 的调度，TQL 函数在传入数据上持续运行。 4. **高级分析**：使用复杂的时序函数，如 `rate()`、`increase()` 和统计聚合。 -### 设置 Source 表 +### 设置 source 表 首先，让我们创建一个 Source 表来存储 HTTP 请求指标： @@ -568,8 +569,10 @@ SELECT * FROM rate_reqs; ```sql DROP FLOW calc_rate; -DROP TABLE http_requests; +DROP FLOW calc_rate_cte; +DROP TABLE http_requests_total; DROP TABLE rate_reqs; +DROP TABLE rate_reqs_cte; ``` ## 下一步 diff --git a/i18n/zh/docusaurus-plugin-content-docs/current/user-guide/flow-computation/expressions.md b/i18n/zh/docusaurus-plugin-content-docs/current/user-guide/flow-computation/expressions.md index 7621d45d4b..b5d97fb314 100644 --- a/i18n/zh/docusaurus-plugin-content-docs/current/user-guide/flow-computation/expressions.md +++ b/i18n/zh/docusaurus-plugin-content-docs/current/user-guide/flow-computation/expressions.md @@ -7,16 +7,14 @@ description: 列出了 GreptimeDB 中 flow 支持的聚合函数和标量函数 ## 聚合函数 -Flow 支持标准 SQL 查询所支持的所有聚合函数，例如 `COUNT`、`SUM`、`MIN`、`MAX` 等。 - -有关详细的函数列表，请参阅 [聚合函数](/reference/sql/functions/df-functions.md#aggregate-functions)。 +Flow 支持 SQL 查询引擎和 Flow 计划所支持的聚合函数，例如 `COUNT`、`SUM`、`MIN` 和 `MAX`。如果查询计划不受支持，Flow 会在创建时失败。有关详细的函数列表，请参阅[聚合函数](/reference/sql/functions/df-functions.md#aggregate-functions)。 ## 标量函数 -Flow 支持标准 SQL 查询所支持的所有标量函数，详见我们的 [SQL 参考](/reference/sql/functions/overview.md)。 +Flow 支持 SQL 查询引擎和 Flow 计划所支持的标量函数。如果查询计划不受支持，Flow 会在创建时失败。详见我们的 [SQL 参考](/reference/sql/functions/overview.md)。 -以下是一些 flow 中最常用的标量函数： +以下是 Flow 中一些常用的标量函数： -- [`date_bin`](/reference/sql/functions/df-functions.md#date_bin): calculate time intervals and returns the start of the interval nearest to the specified timestamp. -- [`date_trunc`](/reference/sql/functions/df-functions.md#date_trunc): truncate a timestamp value to a specified precision. -- [`trunc`](/reference/sql/functions/df-functions.md#trunc): truncate a number to a whole number or truncated to the specified decimal places. \ No newline at end of file +- [`date_bin`](/reference/sql/functions/df-functions.md#date_bin): 计算时间间隔，并返回最接近指定时间戳的区间起点。 +- [`date_trunc`](/reference/sql/functions/df-functions.md#date_trunc): 将时间戳截断到指定精度。 +- [`trunc`](/reference/sql/functions/df-functions.md#trunc): 将数字截断为整数或指定的小数位数。 diff --git a/i18n/zh/docusaurus-plugin-content-docs/current/user-guide/flow-computation/manage-flow.md b/i18n/zh/docusaurus-plugin-content-docs/current/user-guide/flow-computation/manage-flow.md index d86a78ab8c..6476a7505a 100644 --- a/i18n/zh/docusaurus-plugin-content-docs/current/user-guide/flow-computation/manage-flow.md +++ b/i18n/zh/docusaurus-plugin-content-docs/current/user-guide/flow-computation/manage-flow.md @@ -1,5 +1,5 @@ --- -keywords: [创建 flow, 删除 flow, 输入表, sink 表, SQL 语法, 时间窗口, 刷新 flow] +keywords: [创建 flow, 删除 flow, source 表, sink 表, SQL 语法, 时间窗口, 刷新 flow] description: 介绍如何在 GreptimeDB 中创建和删除 flow，包括创建 sink 表、flow 的 SQL 语法和示例。 --- @@ -13,9 +13,10 @@ description: 介绍如何在 GreptimeDB 中创建和删除 flow，包括创建 s Flow 对聚合和 TQL workload 使用 batching mode。简单的非聚合 Flow 查询当前会使用已废弃的 streaming mode，不推荐新 workload 使用。 ::: -## 创建输入表 + +## 创建 source 表 -在创建 `flow` 之前，你需要先创建一张输入表来存储原始的输入数据，比如： +在创建 `flow` 之前，你需要先创建一张 source 表来存储原始数据，比如： ```sql CREATE TABLE temp_sensor_data ( sensor_id INT, @@ -25,29 +26,19 @@ CREATE TABLE temp_sensor_data ( PRIMARY KEY(sensor_id, loc) ); ``` -对于新的 Flow source 表，请避免使用 `WITH ('ttl' = 'instant')`。使用 `ttl='instant'` 的 source 表会回退到已废弃的 streaming mode。请为 source 数据设置合适的 TTL 并保留数据，这样聚合和 TQL Flow workload 才能使用 batching mode。 - -对于已有的旧 streaming-mode 部署，source 表可能会使用 `WITH ('ttl' = 'instant')`： -```sql -CREATE TABLE temp_sensor_data ( - sensor_id INT, - loc STRING, - temperature DOUBLE, - ts TIMESTAMP TIME INDEX, - PRIMARY KEY(sensor_id, loc) -) WITH ('ttl' = 'instant'); -``` - -将 `ttl` 设置为 `'instant'` 会使表立即丢弃插入的数据，并只将行发送给旧 streaming-mode flow 任务。对于新的 workload，该模式已经废弃。 +对于新的 workload，请避免在 Flow source 表上使用 `WITH ('ttl' = 'instant')`。这是旧的使用方式，不推荐用于新的聚合或 TQL workload。请为 source 数据设置合适的保留策略。 ## 创建 sink 表 -flow 把聚合结果写入 sink 表。`CREATE FLOW` 会在该表不存在时自动创建它，因此自己建表是可选的—— -只有需要控制 schema 时才需要自己建，例如指定主键或设置 TTL。无论表由谁创建，它都必须与 flow 查询结果兼容，即： +flow 把聚合结果写入 sink 表。如果 sink 表不存在，`CREATE FLOW` 会在能够从查询结果推断 schema 时自动创建它。 +如果需要控制 schema 或布局，或者查询较复杂难以推断，请预先创建 sink 表。已有 sink 表会根据 flow 查询结果进行校验。 +source 表和 sink 表不能是同一张表。 + +sink 表必须与 flow 查询结果兼容，即： -- **列的顺序和类型**：确保 sink 表中列的顺序和类型与 flow 查询结果匹配。 +- **列的顺序和类型**：对于预先创建的 SQL sink，列的顺序和类型应与查询输出匹配。 - **时间索引**：为 sink 表指定 `TIME INDEX`，通常使用时间窗口函数生成的时间列。 -- **更新时间**：Flow 引擎会自动将更新时间附加到每个计算结果行的末尾。此更新时间存储在 `update_at` 列中。请确保在 sink 表的 schema 中包含此列。 +- **更新时间**：自动创建的 batching SQL sink 会添加 `update_at` 列来记录更新时间。TQL sink 遵循查询输出，不会自动添加 `update_at`。预先创建的 SQL sink 可以与查询输出列数一致，也可以在末尾额外包含一个用于更新时间的时间戳列。 - **Tag**：使用 `PRIMARY KEY` 指定 Tag，与 time index 一起作为行数据的唯一标识，并优化查询性能。 例如： @@ -95,27 +86,31 @@ sink 表包含列 `sensor_id`、`loc`、`max_temp`、`time_window` 和 `update_a CREATE [ OR REPLACE ] FLOW [ IF NOT EXISTS ] SINK TO [ EXPIRE AFTER ] +[ EVAL INTERVAL ] [ COMMENT '' ] [ WITH ( = [, ...]) ] -AS +AS ; ``` +子句必须按上述顺序出现：`EXPIRE AFTER` 在 `EVAL INTERVAL` 之前。 +`EVAL INTERVAL` 会按计划重复执行完整查询。只要 SQL 查询引擎能够生成有效计划，带调度的 SQL Flow +就支持 join、子查询和 SQL CTE。TQL Flow 必须使用 `EVAL INTERVAL`；批处理时间窗口聚合 Flow 可以不使用它。 + 当指定 `OR REPLACE` 时，如果已经存在同名的 flow，它将被更新为新 flow。请注意，这仅影响 flow 任务本身，source 表和 sink 表将不会被更改。当指定 `IF NOT EXISTS` 时，如果 flow 已经存在，它将不执行任何操作，而不是报告错误。还需要注意的是，`OR REPLACE` 不能与 `IF NOT EXISTS` 一起使用。 - `flow-name` 是目录级别的唯一标识符。 - `sink-table-name` 是存储聚合数据的表名。 - 它可以是一个现有的表或一个新表。如果目标表不存在，`flow` 将创建目标表。 - -- `EXPIRE AFTER` 是一个可选的时间间隔，用于从 Flow 引擎中过期数据。 - 有关更多详细信息，请参考 [`EXPIRE AFTER`](#expire-after) 部分。 + 它可以是一个现有的表或一个新表；有关创建和校验行为，请参阅[创建 sink 表](#创建-sink-表)。 +- `EXPIRE AFTER` 是一个可选的时间间隔，用于使 Flow 引擎中的数据过期。有关详细信息，请参考 [`EXPIRE AFTER`](#expire-after) 部分。 +- `EVAL INTERVAL` 是用于按计划执行完整查询的可选时间间隔。TQL Flow 必须使用它。 - `COMMENT` 是 flow 的描述。 - `WITH` 指定 flow 选项。 - 例如，实验性的 `experimental_enable_incremental_read` 选项可以为符合条件的 batching flow 启用增量读取 source 表。 + 本文档介绍的用户 Flow 选项为 `defer_on_missing_source` 和实验性的 `experimental_enable_incremental_read`。 - `SQL` 部分定义了用于持续聚合的查询。 它定义了为 flow 提供数据的源表。 每个 flow 可以有多个源表。 - 有关详细信息，请参考[编写查询](#编写-sql-查询) 部分。 + 有关详细信息，请参考[编写 SQL 查询](#编写-sql-查询)部分。 一个创建 flow 的简单示例： @@ -127,25 +122,21 @@ COMMENT 'My first flow in GreptimeDB' AS SELECT max(temperature) as max_temp, - date_bin('10 seconds'::INTERVAL, ts) as time_window, + date_bin('10 seconds'::INTERVAL, ts) as time_window FROM temp_sensor_data GROUP BY time_window; ``` -创建的 flow 将每 10 秒计算一次 `max(temperature)` 并将结果存储在 `my_sink_table` 中。 -所有在 1 小时内的数据都将用于 flow 计算。 +创建的 flow 会将 `max(temperature)` 按 10 秒时间窗口分组，并将结果存储在 `my_sink_table` 中。 +最近 1 小时内的数据会用于 flow 计算。 ### EXPIRE AFTER `EXPIRE AFTER` 子句指定数据将在 flow 引擎中过期的时间间隔。 -source 表中超出指定过期时间的数据将不再被包含在 flow 的计算范围内。 -同理，sink 表中超过过期时间的历史数据也不会被更新。 -这意味着 flow 引擎在聚合计算时会自动忽略早于该时间间隔的数据。这一机制有助于管理有状态查询（例如涉及 `GROUP BY` 的查询）的状态存储规模。 +对于包含可用时间窗口表达式的 Flow，source 表中早于指定间隔的数据会被排除在计算之外，sink 表中较早的行也不会被更新。这会限制时间窗口 Flow 的状态和重新计算范围，包括涉及 `GROUP BY` 的有状态查询。 -需特别注意的是： -- `EXPIRE AFTER` 子句**不会删除** source 表或 sink 表中的数据，它仅控制 flow 引擎对数据的处理范围 -- 若需删除表数据，请在创建表时通过 [`TTL` 策略](/user-guide/manage-data/overview.md#使用-ttl-策略保留数据)实现 +调度的完整 SQL Flow 和 TQL Flow 会执行未过滤的快照，除非查询本身包含时间谓词；`EXPIRE AFTER` 不会额外添加时间过滤。它不会删除 source 表或 sink 表中的数据。若需删除表数据，请在创建表时通过 [`TTL` 策略](/user-guide/manage-data/overview.md#使用-ttl-策略保留数据)实现。 为 `EXPIRE AFTER` 设置合理的时间间隔，有助于限制 batching 引擎需要向前重新计算结果的时间范围，并避免过度占用资源。它与流处理系统中限制迟到数据范围的机制有相似目的，但新的 Flow workload 应使用 batching mode。 @@ -153,6 +144,19 @@ source 表中超出指定过期时间的数据将不再被包含在 flow 的计 当前时刻若输入数据的 Time Index 超过 1 小时（即早于 09:00:00），则会被判定为过期数据并被忽略。 仅时间戳为 09:00:00 及之后的数据会参与聚合计算，并更新到目标表。 +### 缺少 source 时延迟创建 + +默认情况下，如果任一 source 表不存在，创建 Flow 会失败。将 `defer_on_missing_source` 设置为 `true`， +可以在不失败的情况下持久化一个 pending Flow；当 source 仍未解析时，该 Flow 不会被调度。 + +```sql +CREATE FLOW pending_flow +SINK TO pending_sink +WITH (defer_on_missing_source = 'true') +AS +SELECT * FROM source_created_later; +``` + ### 实验性的增量 source 读取 :::warning 实验性功能 @@ -187,44 +191,32 @@ GROUP BY time_window; ``` -启用该选项后，Flow 会维护每个 region 的 source sequence watermark，并在初始全量快照之后尝试只读取新追加的 source 行。 -这是一个执行优化，不会改变查询结果。 +启用该选项后，Flow 会在初始全量快照之后尝试只读取新追加的 source 行。 +这是一个执行优化，不会改变查询结果，也不构成持久化保证：首次运行、重启后的运行，或无法安全增量读取时，可能会使用全量快照。 当前限制如下： - 所有 source 表都必须是使用 `append_mode = 'true'` 创建的 append-only 表。 如果任意 source 表不是 append-only 表，创建 Flow 会失败。 -- 该优化只适用于 batching SQL flow。 - TQL flow、不支持的聚合形态以及简单的 projection/filter flow 不会使用增量 source 读取。 -- 使用 `ttl = 'instant'` 创建的 source 表当前会使用 streaming 模式，不会使用这个 batching 模式选项。 -- 首次运行仍然需要全量快照。 - 之后的运行在 GreptimeDB 无法安全使用增量 source 读取时，可能会回退到全量快照，或者进行重试/修复。 +- 该优化只适用于符合条件的 batching SQL flow。TQL flow 和不支持增量读取的计划会使用正常的全量快照行为。 ### 编写 SQL 查询 -flow 的 `SQL` 部分类似于标准的 `SELECT` 子句，但有一些不同之处。查询的语法如下： +`AS` 后的 SQL 会作为标准 SQL 查询进行规划。典型的 batching 时间窗口聚合可以使用以下形式： ```sql -SELECT AGGR_FUNCTION(column1, column2,..) [, TIME_WINDOW_FUNCTION() as time_window] FROM GROUP BY {time_window | column1, column2,.. }; +SELECT AGGR_FUNCTION(column1, column2,..) [, TIME_WINDOW_FUNCTION() as time_window] +FROM +GROUP BY {time_window | column1, column2,.. }; ``` -在 `SELECT` 关键字之后只允许以下类型的表达式： -- 聚合函数：有关详细信息，请参阅[表达式](./expressions.md)文档。 -- 时间窗口函数：有关详细信息，请参阅[定义时间窗口](#define-time-window)部分。 -- 标量函数：例如 `col`、`to_lowercase(col)`、`col + 1` 等。这部分与 GreptimeDB 中的标准 `SELECT` 子句相同。 - -查询语法中的其他部分需要注意以下几点： -- 必须包含一个 `FROM` 子句以指定 source 表。由于目前不支持 join 子句，因此只能聚合来自单个表的列。 -- 支持 `WHERE` 和 `HAVING` 子句。`WHERE` 子句在聚合之前过滤数据，而 `HAVING` 子句在聚合之后过滤数据。 -- `DISTINCT` 目前仅适用于 `SELECT DISTINCT column1 ..` 语法。它用于从结果集中删除重复行。`SELECT count(DISTINCT column1) ...` 尚不可用，但将来会添加。 -- `GROUP BY` 子句的工作方式与标准查询相同，即按指定列对数据进行分组，在其中指定时间窗口列对于持续聚合场景至关重要。 - `GROUP BY` 中的其他表达式可以是 literal、列名或 scalar 表达式。 -- 不支持`ORDER BY`、`LIMIT` 和 `OFFSET`。 +具体支持哪些 SQL 表达式和子句取决于 SQL 查询引擎和 Flow 计划。带调度的完整 SQL Flow 支持查询规划器能够生成有效计划的 +join、子查询和 SQL CTE；不受支持的计划会在创建 Flow 时失败。对于 batching 时间窗口聚合，`GROUP BY` 通常包含时间窗口表达式。 +有关 Flow 查询中常用的函数，请参阅[表达式](./expressions.md)；有关固定时间窗口，请参阅[定义时间窗口](#define-time-window)。 有关如何在实时分析、监控和仪表板中使用持续聚合的更多示例，请参阅[持续聚合](./continuous-aggregation.md)。 - ### 定义时间窗口 时间窗口是持续聚合查询的重要属性。 @@ -253,9 +245,25 @@ GROUP BY time_window; 请参阅 [`date_bin`](/reference/sql/functions/df-functions.md#date_bin)。 :::tip 提示 -目前，flow 依赖时间窗口表达式来确定如何增量更新结果。因此，建议尽可能使用相对较小的时间窗口。 +时间窗口表达式可帮助 Flow 确定如何增量更新结果。合适的窗口大小取决于 workload 和查询语义。 ::: +## 检查 Flow + +可以使用以下语句和系统表检查 Flow 的定义及运行时信息： + +```sql +SHOW FLOWS; +SHOW CREATE FLOW my_flow; +SHOW FLOW STATUS LIKE 'my%'; +SELECT * FROM information_schema.flows; +SELECT * FROM information_schema.flow_statistics; +``` + +`SHOW FLOWS` 列出 Flow，`SHOW CREATE FLOW` 返回 Flow 定义，`SHOW FLOW STATUS` 返回运行时统计信息。 +`information_schema` 中的两张表分别提供定义和统计信息。在分布式部署中，运行时字段初始可能为 `NULL`， +其值也可能落后于最新状态。 + ## 刷新 flow 当 source 表中有新数据到达时，flow 引擎会在短时间内（比如数秒）自动处理聚合操作。 diff --git a/i18n/zh/docusaurus-plugin-content-docs/current/user-guide/flow-computation/overview.md b/i18n/zh/docusaurus-plugin-content-docs/current/user-guide/flow-computation/overview.md index 77f45872d9..6055c02c57 100644 --- a/i18n/zh/docusaurus-plugin-content-docs/current/user-guide/flow-computation/overview.md +++ b/i18n/zh/docusaurus-plugin-content-docs/current/user-guide/flow-computation/overview.md @@ -7,7 +7,7 @@ description: 了解 GreptimeDB 的 Flow 引擎如何对持续写入的数据进 GreptimeDB 的 Flow 引擎可以对持续写入的数据进行实时计算。 它特别适用于提取 - 转换 - 加载 (ETL) 过程，或执行持续聚合，例如求和、平均值和其他时间窗口计算。 -每写入一行数据就增量更新 sink 表，聚合在写入时完成，而不是每次查询时重算。 +Flow 会在处理 source 数据时将计算结果物化到 sink 表中，因此查询可以直接读取计算结果，而不必从原始数据重新计算。 使用案例包括： @@ -20,10 +20,8 @@ Flow 对聚合和 TQL workload 使用 batching mode。简单的非聚合 Flow ## 程序模型 -在将数据插入 source 表后， -数据会同时被写入到 Flow 引擎中。 -在每个触发间隔（一秒）时， -Flow 引擎执行指定的计算并将结果更新到 sink 表中。 +在将数据插入 source 表后，数据会提供给 Flow 引擎处理。 +Flow 随后执行指定的计算并将结果更新到 sink 表中。 source 表和 sink 表都是 GreptimeDB 中的时间序列表。 在创建 Flow 之前， 定义这些表的 schema 并设计 Flow 以指定计算逻辑是至关重要的。