Skip to content

Commit 617dc51

Browse files
BohuTANGBohuTANG
authored
docs: refresh continuous pipeline demos (#2948)
* docs: refresh continuous pipeline demos * docs: polish stream/task examples * docs: streamline continuous pipeline index * docs(cn): refresh continuous pipeline guides * docs(cn): clarify task naming * docs(stream): simplify example titles * docs(stream): shorten example title * docs(stream): clarify example headings * docs: streamline stream/task intros * docs(cn): polish stream/task wording * docs(cn): keep english terms in stream example * docs(cn): simplify step headings * docs(task): align stream schema in guard
1 parent 409a32e commit 617dc51

File tree

6 files changed

+634
-1052
lines changed

6 files changed

+634
-1052
lines changed

‎docs/cn/guides/40-load-data/05-continuous-data-pipelines/01-stream.md‎

Lines changed: 148 additions & 382 deletions
Large diffs are not rendered by default.

‎docs/cn/guides/40-load-data/05-continuous-data-pipelines/02-task.md‎

Lines changed: 116 additions & 131 deletions
Original file line numberDiff line numberDiff line change
@@ -1,169 +1,154 @@
11
---
2-
title: 使用任务自动化数据加载
3-
sidebar_label: 任务
2+
title: 使用任务（Task）自动化数据加载
3+
sidebar_label: 任务（Task）
44
---
55

6-
任务封装了特定的 SQL 语句，设计用于按预定间隔执行、由特定事件触发或作为任务序列的一部分执行。在 Databend Cloud 中，任务通常用于定期捕获流中的变更数据（例如新增记录），然后将这些数据同步到指定目标。此外，任务还支持 [Webhook](https://en.wikipedia.org/wiki/Webhook) 等消息系统，可按需发送错误消息和通知。
6+
Task 是“把 SQL 交给 Databend 代跑”的方式。你可以让它按固定频率运行、在另一任务结束后运行，或者在某个 Stream 报告有增量时再运行。下面先看定义 Task 时需要关注的几个开关，再通过两个动手示例理解它如何和 Stream 配合。
77

8-
## 创建任务
8+
## Task 构建要素
99

10-
本主题详细说明在 Databend Cloud 中创建任务的步骤。您可以使用 [CREATE TASK](/sql/sql-commands/ddl/task/ddl-create_task) 命令创建任务。创建任务时，请参照下图设计工作流：
10+
- **名称与计算仓库** – 每个 Task 都需要一个 Warehouse。
11+
```sql
12+
CREATE TASK ingest_orders
13+
WAREHOUSE = 'etl_wh'
14+
AS SELECT 1;
15+
```
16+
- **触发方式** – `SCHEDULE = 2 MINUTE`、CRON，或 `AFTER <task>`（适用于 DAG）。
17+
- **执行条件** – `WHEN STREAM_STATUS('mystream') = TRUE` 这类布尔表达式，只有条件满足才运行。
18+
- **错误策略** – `SUSPEND_TASK_AFTER_NUM_FAILURES`、`ERROR_INTEGRATION` 等参数可在失败多次后暂停并发通知。
19+
- **SQL 负载** – `AS` 后的内容就是 Task 要执行的语句，可以是一条 INSERT/COPY/MERGE，也可以是 BEGIN...END。
1120

12-
![alt text](/img/load/task.png)
21+
## 示例 1：定时 COPY
1322

14-
1. 为任务设置名称。
15-
2. 指定运行任务的计算集群。如需创建计算集群，请参阅 [使用计算集群](/guides/cloud/using-databend-cloud/warehouses)。
16-
3. 确定任务触发方式。
23+
持续生成 Parquet 并导入表。记得把 `'etl_wh_small'` 换成你自己的 Warehouse。
1724

18-
- 可通过指定分钟或秒为间隔来调度任务，或使用 CRON 表达式配合可选时区实现更精确的调度。
25+
### 步骤 1： 准备演示对象
1926

20-
```sql title='示例:'
21-
-- 此任务每 2 分钟运行一次
22-
CREATE TASK mytask
23-
WAREHOUSE = 'default'
24-
// highlight-next-line
25-
SCHEDULE = 2 MINUTE
26-
AS ...
27+
```sql
28+
CREATE DATABASE IF NOT EXISTS task_demo;
29+
USE task_demo;
30+
31+
CREATE OR REPLACE TABLE sensor_events (
32+
event_time TIMESTAMP,
33+
sensor_id INT,
34+
temperature DOUBLE,
35+
humidity DOUBLE
36+
);
2737
28-
-- 此任务每天午夜（东京时间）在 Asia/Tokyo 时区运行
29-
CREATE TASK mytask
30-
WAREHOUSE = 'default'
31-
// highlight-next-line
32-
SCHEDULE = USING CRON '0 0 0 * * *' 'Asia/Tokyo'
33-
AS ...
38+
CREATE OR REPLACE STAGE sensor_events_stage;
3439
```
3540

36-
- 或者，您可以在任务间建立依赖关系，将任务设置为 [有向无环图](https://en.wikipedia.org/wiki/Directed_acyclic_graph) 中的子任务。
41+
### 步骤 2： Task 1 —— 生成文件
3742

38-
```sql title='示例:'
39-
-- 此任务依赖于 DAG 中 'task_root' 任务的完成
40-
CREATE TASK mytask
41-
WAREHOUSE = 'default'
42-
// highlight-next-line
43-
AFTER task_root
44-
AS ...
43+
```sql
44+
CREATE OR REPLACE TASK task_generate_data
45+
WAREHOUSE = 'etl_wh_small'
46+
SCHEDULE = 1 MINUTE
47+
AS
48+
COPY INTO @sensor_events_stage
49+
FROM (
50+
SELECT
51+
NOW() AS event_time,
52+
number AS sensor_id,
53+
20 + RAND() * 5 AS temperature,
54+
60 + RAND() * 10 AS humidity
55+
FROM numbers(100)
56+
)
57+
FILE_FORMAT = (TYPE = PARQUET);
4558
```
4659

47-
4. 指定任务执行条件，允许基于布尔表达式控制任务执行（可选）。
60+
### 步骤 3： Task 2 —— 将文件导入表
4861

49-
```sql title='示例:'
50-
-- 此任务每 2 分钟运行一次，仅当 'mystream' 包含数据变更时执行 AS 后的 SQL
51-
CREATE TASK mytask
52-
WAREHOUSE = 'default'
53-
SCHEDULE = 2 MINUTE
54-
// highlight-next-line
55-
WHEN STREAM_STATUS('mystream') = TRUE
56-
AS ...
62+
```sql
63+
CREATE OR REPLACE TASK task_consume_data
64+
WAREHOUSE = 'etl_wh_small'
65+
SCHEDULE = 1 MINUTE
66+
AS
67+
COPY INTO sensor_events
68+
FROM @sensor_events_stage
69+
PATTERN = '.*[.]parquet'
70+
FILE_FORMAT = (TYPE = PARQUET)
71+
PURGE = TRUE;
5772
```
5873

59-
5. 指定任务出错时的处理方式，包括设置连续失败次数以暂停任务，以及指定错误通知的集成方式。有关设置错误通知的更多信息，请参阅 [配置通知集成](#configuring-notification-integrations)。
60-
61-
```sql title='示例:'
62-
-- 此任务将在连续失败 3 次后暂停
63-
CREATE TASK mytask
64-
WAREHOUSE = 'default'
65-
// highlight-next-line
66-
SUSPEND_TASK_AFTER_NUM_FAILURES = 3
67-
AS ...
68-
69-
-- 此任务将使用 'my_webhook' 集成发送错误通知
70-
CREATE TASK mytask
71-
WAREHOUSE = 'default'
72-
// highlight-next-line
73-
ERROR_INTEGRATION = 'my_webhook'
74-
AS ...
74+
### 步骤 4： 恢复 Task
75+
76+
```sql
77+
ALTER TASK task_generate_data RESUME;
78+
ALTER TASK task_consume_data RESUME;
7579
```
7680

77-
6. 指定任务将执行的 SQL 语句。
81+
### 步骤 5： 观察运行情况
7882

79-
```sql title='示例:'
80-
-- 此任务每年更新 'employees' 表中的 'age' 列，使其递增 1
81-
CREATE TASK mytask
82-
WAREHOUSE = 'default'
83-
SCHEDULE = USING CRON '0 0 1 1 * *' 'UTC'
84-
// highlight-next-line
85-
AS
86-
UPDATE employees
87-
SET age = age + 1;
83+
```sql
84+
SHOW TASKS LIKE 'task_%';
85+
LIST @sensor_events_stage;
86+
SELECT * FROM sensor_events ORDER BY event_time DESC LIMIT 5;
87+
SELECT * FROM task_history('task_consume_data', 5);
8888
```
8989

90-
## 查看已创建的任务
91-
92-
要查看组织创建的所有任务，请登录 Databend Cloud 并转到 **数据** > **任务**。您可以查看每个任务的详细信息，包括状态和调度计划。
93-
94-
要查看任务运行历史记录，请转到 **监控** > **任务历史**。您可以查看每次任务运行的结果、完成时间等详细信息。
95-
96-
## 配置通知集成
97-
98-
Databend Cloud 允许您为任务配置错误通知，在任务执行出错时自动发送通知。当前支持 Webhook 集成，可实现错误事件与外部系统或服务的实时通信。
99-
100-
### 任务错误负载
101-
102-
任务错误负载指任务执行出错时作为错误通知发送的数据或信息。该负载通常包含错误详情，如错误代码、错误消息、时间戳以及其他有助于诊断和解决问题的上下文信息。
103-
104-
```json title='任务错误负载示例:'
105-
{
106-
"version": "1.0",
107-
"messageId": "063e40ab-0b55-439e-9cd2-504c496e1566",
108-
"messageType": "TASK_FAILED",
109-
"timestamp": "2024-03-19T02:37:21.160705788Z",
110-
"tenantId": "tn78p61xz",
111-
"taskName": "my_task",
112-
"taskId": "15",
113-
"rootTaskName": "my_task",
114-
"rootTaskId": "15",
115-
"messages": [
116-
{
117-
"runId": "unknown",
118-
"scheduledTime": "2024-03-19T02:37:21.157169855Z",
119-
"queryStartTime": "2024-03-19T02:37:21.043090475Z",
120-
"completedTime": "2024-03-19T02:37:21.157169205Z",
121-
"queryId": "88bb9d5d-5d5e-4e52-92cc-b1953406245a",
122-
"errorKind": "UnexpectedError",
123-
"errorCode": "500",
124-
"errorMessage": "query sync failed: All attempts fail:\n#1: query error: code: 1006, message: divided by zero while evaluating function `divide(1, 0)`"
125-
}
126-
]
127-
}
90+
### 步骤 6： 调整或改写 Task
91+
92+
```sql
93+
ALTER TASK task_consume_data
94+
SET SCHEDULE = 30 SECOND,
95+
WAREHOUSE = 'etl_wh_medium';
96+
97+
ALTER TASK task_consume_data
98+
MODIFY AS
99+
COPY INTO sensor_events
100+
FROM @sensor_events_stage
101+
FILE_FORMAT = (TYPE = PARQUET);
102+
103+
ALTER TASK task_consume_data RESUME;
104+
105+
SELECT *
106+
FROM task_history('task_consume_data', 5)
107+
ORDER BY completed_time DESC;
128108
```
129109

130-
### 使用示例
110+
## 示例 2：Stream 条件 Task
131111

132-
在为任务配置错误通知前，您需要使用 [CREATE NOTIFICATION INTEGRATION](/sql/sql-commands/ddl/notification/ddl-create-notification) 命令创建通知集成。以下示例展示了如何为任务创建和配置通知集成。该示例使用 [Webhook.site](http://webhook.site) 模拟消息系统，接收来自 Databend Cloud 的负载。
112+
只有当 Stream 报告“有增量”时才运行，避免空跑。
133113

134-
1. 在浏览器中打开 [Webhook.site](http://webhook.site)，获取您的 Webhook URL。
114+
### 步骤 1： 创建 Stream 与结果表
135115

136-
![alt text](/img/load/webhook-1.png)
116+
```sql
117+
CREATE OR REPLACE STREAM sensor_events_stream
118+
ON TABLE sensor_events
119+
APPEND_ONLY = false;
120+
121+
CREATE OR REPLACE TABLE sensor_events_latest AS
122+
SELECT *
123+
FROM sensor_events
124+
WHERE 1 = 0;
125+
```
137126

138-
2. 在 Databend Cloud 中创建通知集成，然后创建带通知集成的任务：
127+
### 步骤 2： 定义条件 Task
139128

140129
```sql
141-
-- 创建名为 'my_task' 的任务，每分钟运行一次，错误通知发送至 'my_webhook'
142-
-- 故意除以零以生成错误
143-
CREATE TASK my_task
144-
WAREHOUSE = 'default'
145-
SCHEDULE = 1 MINUTE
146-
ERROR_INTEGRATION = 'my_webhook'
130+
CREATE OR REPLACE TASK task_stream_merge
131+
WAREHOUSE = 'etl_wh_small'
132+
SCHEDULE = 1 MINUTE
133+
WHEN STREAM_STATUS('task_demo.sensor_events_stream') = TRUE
147134
AS
148-
SELECT 1 / 0;
149-
150-
-- 创建名为 'my_webhook' 的通知集成，用于发送 webhook 通知
151-
CREATE NOTIFICATION INTEGRATION my_webhook
152-
TYPE = WEBHOOK
153-
ENABLED = TRUE
154-
WEBHOOK = (
155-
url = '<YOUR-WEBHOOK_URL>',
156-
method = 'POST'
157-
);
135+
INSERT INTO sensor_events_latest
136+
SELECT *
137+
FROM sensor_events_stream;
158138
159-
-- 创建后恢复任务
160-
ALTER TASK my_task RESUME;
139+
ALTER TASK task_stream_merge RESUME;
161140
```
162141

163-
3. 稍等片刻，您将看到 webhook 开始接收来自创建的任务的负载。
142+
### 步骤 3： 查看增量与历史
164143

165-
![alt text](/img/load/webhook-2.png)
144+
```sql
145+
SELECT *
146+
FROM sensor_events_latest
147+
ORDER BY event_time DESC
148+
LIMIT 5;
166149
167-
## 使用示例
150+
SELECT *
151+
FROM task_history('task_stream_merge', 5);
152+
```
168153

169-
完整演示如何通过流捕获数据变更并使用任务同步，请参阅 [示例：实时跟踪和转换数据](01-stream.md#example-tracking-and-transforming-data-in-real-time)。
154+
只要 `STREAM_STATUS('<database>.<stream_name>')` 返回 TRUE（例如 `task_demo.sensor_events_stream`），Task 就会运行；否则保持暂停，直到下一批增量到达。

‎docs/cn/guides/40-load-data/05-continuous-data-pipelines/index.md‎

Lines changed: 15 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -2,26 +2,24 @@
22
title: 持续数据管道
33
---
44

5-
## 数据管道简介
5+
在 Databend 中构建 CDC（Change Data Capture）流程只需两种原语：
66

7-
数据管道能够自动化地将来自不同来源的数据移动并转换至 Databend。它们确保数据流畅传输，对于快速且持续地处理和分析数据至关重要。
7+
- **Stream**：捕获表的 INSERT/UPDATE/DELETE，一直保留到被消费。
8+
- **Task**：按计划或在 Stream 有新数据时自动运行 SQL。
89

9-
在持续数据管道中，一项名为 **变更数据捕获 (CDC)** 的特殊功能发挥着关键作用。借助 Databend，CDC 变得简单高效，仅需通过 Streams 和 Tasks 执行几条简单命令即可实现。
10+
## 快速入口
1011

11-
## 理解变更数据捕获 (CDC)
12+
- [示例 1：仅追加 Stream](./01-stream.md#示例-1仅追加-stream) – 捕获插入并写入目标表。
13+
- [示例 2：标准 Stream](./01-stream.md#示例-2标准-stream含-update-delete) – 了解更新、删除在 Stream 中的表现。
14+
- [示例 3：增量 Join](./01-stream.md#示例-3增量-join--计算) – 使用 `WITH CONSUME` 做批式增量聚合。
15+
- [示例 1：定时 COPY 任务](./02-task.md#示例-1定时-copy) – 两个任务生成并导入 Parquet。
16+
- [示例 2：Stream 条件任务](./02-task.md#示例-2基于-stream-的条件任务) – 只有 Stream 有增量时才触发。
1217

13-
CDC 是指流对象捕获应用于数据库表的插入、更新和删除操作的过程。它包含有关每次变更的元数据，从而能够基于修改后的数据执行操作。Databend 中的 CDC 在源表中跟踪行级变更，创建一个"变更表"来反映两个事务时间点之间的数据修改。
18+
## 为什么选择 Databend CDC
1419

15-
## 使用变更数据捕获 (CDC) 的优势
20+
- **轻量**：Stream 只保存尚未消费的增量。
21+
- **事务一致**：消费 Stream 的语句成功才会清空，失败即回滚。
22+
- **增量友好**：配合 `WITH CONSUME` 能多次运行同一 SQL，每次只处理新数据。
23+
- **自动化**：Task 让任何 SQL 都能定时/触发执行。
1624

17-
1. **快速实时数据加载**：优化来自事务数据库的实时数据加载流程，几乎可在秒级完成。
18-
2. **不影响原始数据**：安全可靠，不会损坏数据或其来源系统。
19-
3. **克服批量 ETL 的局限性**：超越传统的批量 ETL 方法，后者对于持续数据更新而言速度较慢且效率较低。
20-
21-
## Databend 持续数据管道的核心特性
22-
23-
Databend 通过以下特性增强了持续数据管道：
24-
25-
- **持续数据跟踪与转换**：支持数据的实时跟踪与转换。[通过 Streams 了解数据跟踪与转换的更多信息](./01-stream.md)。
26-
27-
- **循环任务**：支持调度和管理循环数据处理任务，确保数据管道的高效性和可靠性。该功能目前处于私有预览阶段。
25+
先完成 Stream 示例，再组合 Task，即可搭建自己的持续数据管道。

0 commit comments

Comments
 (0)