当前位置：首页 > news >正文

2025-04-07（DS复习）：Databricks DLT 详解

news 来源：原创 2025/6/1 6:59:48

Databricks Delta Live Tables (DLT) 详解

Delta Live Tables (DLT) 是 Databricks 提供的一个智能框架，用于构建可靠、可扩展的数据处理管道。它简化了ETL(提取、转换、加载)和ELT(提取、加载、转换)流程的开发和管理，特别适合在数据湖house架构中实现高质量的数据处理。

1. DLT 核心概念

1.1 基本定义

DLT 是一个声明式的数据管道框架，它：

基于 Delta Lake 构建
提供表级别的抽象
自动管理任务依赖关系和执行顺序
内置数据质量监控功能

1.2 关键特性

声明式编程模型：

开发者只需定义数据处理逻辑(使用SQL或Python)
系统自动处理执行计划、依赖管理和任务调度

自动运维能力：

自动优化(文件合并、Z-ordering等)
自动错误处理和重试机制
内置数据质量验证(通过EXPECTATIONS)

统一批流处理：

同一套代码可以处理批量和流式数据
自动增量处理

2. DLT 架构组件

2.1 管道(Pipeline)

数据处理作业的逻辑单元
包含源数据、转换逻辑和目标表的完整定义
可以配置触发方式(手动、连续或按计划)

2.2 表类型

Live Tables：物化视图，存储实际数据
Streaming Live Tables：处理流式数据的物化视图
Views：虚拟表，不存储数据

2.3 数据质量框架

通过@expect装饰器或EXPECT关键字定义数据质量规则
可配置违规处理方式(记录、丢弃或失败)

3. DLT 工作方式

3.1 开发流程

定义源数据
编写转换逻辑(SQL或Python)
添加数据质量约束
部署管道
监控和维护

3.2 执行模式

Triggered：按需运行(批处理)
Continuous：持续运行(流处理)

3.3 更新策略

Full Refresh：完全重新计算所有表
Incremental：只处理新数据(自动检测)

4. DLT 代码示例

4.1 Python 示例

python

复制

from dlt import *

# 定义源数据
@table
def raw_sales():
    return spark.read.format("json").load("/data/sales/raw")

# 定义转换
@table
@expect("valid amount", "amount > 0")
def cleaned_sales():
    return spark.sql("""
        SELECT 
            order_id, 
            customer_id, 
            amount, 
            date 
        FROM LIVE.raw_sales 
        WHERE amount IS NOT NULL
    """)

# 定义聚合
@table
def sales_by_customer():
    return spark.sql("""
        SELECT 
            customer_id, 
            SUM(amount) as total_spent,
            COUNT(*) as order_count
        FROM LIVE.cleaned_sales
        GROUP BY customer_id
    """)

4.2 SQL 示例

sql

复制

-- 创建源表
CREATE LIVE TABLE raw_events
COMMENT "原始事件数据"
AS SELECT * FROM json.`/data/events/raw`;

-- 创建清洗后的表
CREATE LIVE TABLE cleaned_events (
  CONSTRAINT valid_timestamp EXPECT (timestamp IS NOT NULL) ON VIOLATION DROP ROW
COMMENT "清洗后的事件数据"
AS SELECT 
    user_id,
    event_type,
    CAST(timestamp AS TIMESTAMP) AS event_time
FROM LIVE.raw_events;

-- 创建聚合表
CREATE LIVE TABLE daily_event_counts
COMMENT "每日事件统计"
AS SELECT
    DATE(event_time) AS day,
    event_type,
    COUNT(*) AS event_count
FROM LIVE.cleaned_events
GROUP BY day, event_type;