Volume + Pipe + Dynamic Table 端到端实践

"数据湖加速"是指在不搬迁数据的前提下，通过对象存储挂载（Volume）、持续数据导入（Pipe）和增量计算（Dynamic Table）三大能力，用 Serverless 算力直接查询、加工和消费对象存储中的文件数据，替代传统 Spark/Hive ETL 和 Presto/Trino Ad hoc 查询。

适用场景：

• 文件自动入库：定期上传到 OSS/COS/S3 的 CSV/Parquet 文件，Pipe 自动检测并入库，无需人工触发
• 增量 ETL：文件入库后，Dynamic Table 自动增量计算聚合指标，T+1 报表生成无延迟
• 存量数据激活：已有大量历史文件在对象存储中，通过 Volume 挂载即可直接查询，无需数据迁移

核心数据流：

OSS/COS/S3 文件 → External Volume (挂载) → Pipe (持续导入) → 目标表 → Dynamic Table (增量聚合) ↕ ↕ COPY INTO/SELECT FROM COPY INTO/SELECT FROM

核心概念

三者配合形成一个 自驱动数据管道：文件上传 → 自动入库 → 自动聚合，全程无需人工调度。

涉及的 SQL 命令

CREATE STORAGE CONNECTION

CREATE EXTERNAL VOLUME

COPY INTO VOLUME

SELECT FROM VOLUME

DIRECTORY()

ALTER VOLUME REFRESH

AUTO_REFRESH=FALSE

CREATE PIPE

ALTER PIPE

DESC PIPE EXTENDED

load_history()

CREATE DYNAMIC TABLE

REFRESH DYNAMIC TABLE

SHOW DYNAMIC TABLE REFRESH HISTORY

前置准备

以下以阿里云 OSS 为例，使用

semantic_model_test

DEFAULT

端到端实践

步骤一：创建 Storage Connection

建立 Lakehouse 与 OSS 之间的认证通道。

-- 创建 OSS 存储连接 CREATE STORAGE CONNECTION IF NOT EXISTS my_oss_conn TYPE OSS access_id = '<your_access_key_id>' access_key = '<your_access_key_secret>' ENDPOINT = 'oss-cn-shanghai.aliyuncs.com';

access_id

access_key

ACCESS_KEY_ID

ACCESS_KEY_SECRET

ACCESS_KEY

SECRET_KEY

步骤二：创建外部 Volume

将 OSS Bucket 子目录挂载为 Lakehouse Volume。

CREATE EXTERNAL VOLUME IF NOT EXISTS my_data_vol LOCATION 'oss://my-bucket/data/' USING CONNECTION my_oss_conn DIRECTORY = (ENABLE = TRUE, AUTO_REFRESH = FALSE) RECURSIVE = TRUE COMMENT '数据湖加速专用 Volume';

关键参数说明：

LOCATION

USING CONNECTION

DIRECTORY.ENABLE

DIRECTORY()

AUTO_REFRESH

TRUE

FALSE

ALTER VOLUME REFRESH

RECURSIVE

步骤三：创建源表并导出到 Volume

验证 Volume 的双向读写能力。

-- 1. 创建源表并插入测试数据 CREATE TABLE IF NOT EXISTS sales_source ( id BIGINT COMMENT '订单ID', product STRING COMMENT '商品名', category STRING COMMENT '品类', amount DECIMAL(10,2) COMMENT '金额', dt STRING COMMENT '日期' ) COMMENT '数据湖加速测试源表'; INSERT INTO sales_source VALUES (1, 'iPhone 15', 'Electronics', 8999.00, '2026-06-01'), (2, 'MacBook Pro', 'Electronics', 14999.00, '2026-06-01'), (3, 'AirPods', 'Electronics', 1299.00, '2026-06-01'), (4, 'Nike Air Max', 'Sports', 899.00, '2026-06-01'), (5, 'Yoga Mat', 'Sports', 199.00, '2026-06-01'); -- 2. 导出为 CSV 到 Volume COPY INTO VOLUME my_data_vol SUBDIRECTORY 'export/' FROM TABLE sales_source FILE_FORMAT = (TYPE = CSV); -- 3. 导出为 Parquet 到 Volume COPY INTO VOLUME my_data_vol SUBDIRECTORY 'export/' FROM TABLE sales_source FILE_FORMAT = (TYPE = PARQUET);

COPY INTO VOLUME

SUBDIRECTORY

Syntax error at or near 'FROM'

SUBDIRECTORY '/'

COPY INTO VOLUME

FILE_FORMAT = (TYPE = CSV/PARQUET)

USING CSV

USING

SELECT FROM VOLUME

步骤四：验证 Volume 文件读写

-- 刷新目录缓存（AUTO_REFRESH=FALSE 时需手动刷新） ALTER VOLUME my_data_vol REFRESH; -- 查看导出的文件 SELECT relative_path, size, last_modified_time FROM DIRECTORY(VOLUME my_data_vol) WHERE relative_path LIKE 'export/%'; -- 直接查询 CSV 文件 SELECT * FROM VOLUME my_data_vol USING CSV FILES('export/part00001.csv'); -- 直接查询 Parquet 文件（保留列名） SELECT id, product, category, amount, dt FROM VOLUME my_data_vol USING PARQUET FILES('export/part00001.parquet');

f0, f1, f2...

OPTIONS('header'='true')

步骤五：创建 Pipe 持续导入

Pipe 持续监控 Volume 中的新文件，自动导入到目标表。

-- 1. 为 Pipe 创建专用 Volume（必须指向独立子目录） CREATE EXTERNAL VOLUME IF NOT EXISTS pipe_vol LOCATION 'oss://my-bucket/data/incoming/' USING CONNECTION my_oss_conn DIRECTORY = (ENABLE = TRUE, AUTO_REFRESH = TRUE) RECURSIVE = TRUE COMMENT 'Pipe 持续导入专用 Volume'; -- 2. 创建目标表 CREATE TABLE IF NOT EXISTS sales_ods ( id BIGINT COMMENT '订单ID', product STRING COMMENT '商品名', category STRING COMMENT '品类', amount DECIMAL(10,2) COMMENT '金额', dt STRING COMMENT '日期' ) COMMENT 'ODS 层——Pipe 导入目标表'; -- 3. 创建 Pipe（LIST_PURGE 模式） CREATE PIPE IF NOT EXISTS sales_pipe INGEST_MODE = 'LIST_PURGE' VIRTUAL_CLUSTER = 'DEFAULT' COMMENT '销售数据持续导入管道' AS COPY INTO sales_ods FROM VOLUME pipe_vol USING CSV PURGE = TRUE;

Pipe 管理

-- 查看 Pipe 状态 DESC PIPE EXTENDED sales_pipe; -- 关键字段：pipe_status (RUNNING/PAUSED)、ingest_mode、input_name、output_name -- 暂停 Pipe（停止扫描新文件） ALTER PIPE sales_pipe SET PIPE_EXECUTION_PAUSED = TRUE; -- 恢复 Pipe（重新开始扫描） ALTER PIPE sales_pipe SET PIPE_EXECUTION_PAUSED = FALSE; -- 查看已导入的文件记录（7 天保留期） SELECT * FROM load_history('sales_ods'); -- 返回：file_path、last_copy_time、file_size、status、first_error_message

load_history

触发 Pipe 加载

Pipe 创建后立即开始运行（约 30 秒轮询一次）。向 Volume 路径写入新文件即可触发加载：

-- 通过 COPY INTO VOLUME 模拟"新文件到达" COPY INTO VOLUME pipe_vol SUBDIRECTORY '/' FROM (SELECT * FROM sales_source WHERE dt = '2026-06-01') FILE_FORMAT = (TYPE = CSV); -- 稍等片刻后验证数据已加载 SELECT COUNT(*) FROM sales_ods; -- 应返回 5

步骤六：创建 Dynamic Table 增量消费

基于 Pipe 导入后的表，创建 Dynamic Table 实现自动增量聚合。

-- 开启源表变更跟踪（增量刷新前提） ALTER TABLE sales_ods SET PROPERTIES ('change_tracking' = 'true'); -- 创建 Dynamic Table，按品类汇总 CREATE OR REPLACE DYNAMIC TABLE sales_summary REFRESH INTERVAL 1 HOUR vcluster DEFAULT COMMENT '按品类汇总——增量刷新' AS SELECT category, COUNT(*) AS order_cnt, SUM(amount) AS total_amount, AVG(amount) AS avg_amount, MIN(dt) AS min_date, MAX(dt) AS max_date FROM sales_ods GROUP BY category; -- 立即触发首次刷新（重置刷新基准时间） REFRESH DYNAMIC TABLE sales_summary; -- 查询结果 SELECT * FROM sales_summary ORDER BY category;

REFRESH INTERVAL 1 HOUR

REFRESH

查看 DT 刷新历史

SHOW DYNAMIC TABLE REFRESH HISTORY WHERE name = 'sales_summary'; -- 关注字段：state (SUCCEED)、refresh_mode (INCREMENTAL/FULL)、duration、source_tables

完整数据流验证

-- 验证各环节数据一致性 SELECT 'Source' AS stage, COUNT(*) AS rows FROM sales_source UNION ALL SELECT 'ODS' AS stage, COUNT(*) AS rows FROM sales_ods UNION ALL SELECT 'Summary' AS stage, COUNT(*) AS rows FROM sales_summary;

最佳实践

文件大小建议

Volume 与 Pipe 设计原则

1. 每个 Pipe 独立 Volume：不同 Pipe 不能共用同一 Volume，避免互相干扰
2. Volume 指向子目录：不要指向 bucket 根路径，否则 Pipe 创建报错
3. LIST_PURGE vs EVENT_NOTIFICATION：

   • LIST_PURGE

     LIST_PURGE
     ：配置简单，适合大多数场景，加载后删除源文件

   • EVENT_NOTIFICATION

     EVENT_NOTIFICATION
     ：低延迟，保留源文件，但仅支持 OSS+S3，且需要额外配置 MNS/SQS

Dynamic Table 设计原则

1. 使用 GP 型 VCluster（如

   DEFAULT

   DEFAULT
   ）：GP 型支持小文件合并，AP 型不支持
2. 开启 change_tracking：源表未开启时 DT 每次全量刷新，无法增量
3. 创建后立即 REFRESH：确保首次数据可用，同时重置刷新基准时间

数据生命周期

文件上传 → Pipe 扫描 → COPY INTO 入库 → PURGE 删除 → Dynamic Table 增量刷新 ↓ ↓ ↓ ↓ ↓ OSS 写入 30 秒轮询 load_history 记录 源文件删除 聚合更新

测试验证结果

以下结果来自阿里云上海实例（

f8866243

AUTO_REFRESH=FALSE

SUBDIRECTORY

注意事项

COPY INTO VOLUME

SUBDIRECTORY

SUBDIRECTORY '/'

OPTIONS('header'='true')

AUTO_REFRESH=FALSE

ALTER VOLUME name REFRESH

load_history

last_copy_time

last_load_time

DEFAULT

相关文档

• 多云统一数据湖加速方案 — 阿里云/腾讯云/AWS 三云实测对比
• Volume 概览 — Volume 概念、类型和文件操作
• 对象存储 Pipe — LIST_PURGE 和 EVENT_NOTIFICATION 完整配置
• Pipe 概览 — Pipe 与 Table Stream 对比
• Dynamic Table 概览 — 增量计算机制
• 创建外部 Volume — 完整 DDL 语法
• 从 Volume 导入数据 — COPY INTO 语法
• 导出数据到 Volume — 导出语法
• 查询 SHOW JOBS — 按 query_tag 过滤 Pipe 作业