现代数据湖架构设计：从存储到分析的完整方案

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180

# Delta Lake数据湖实现
from delta import DeltaTable
from pyspark.sql import SparkSession
from pyspark.sql.functions import *

class DataLakeBuilder:
    def __init__(self, spark: SparkSession, base_path: str):
        self.spark = spark
        self.base_path = base_path

    def build_bronze_layer(self, source_system: str):
        """
        Bronze层：原始数据层
        - 保持原始格式
        - 不做任何转换
        - 用于审计和重放
        """
        bronze_path = f"{self.base_path}/bronze/{source_system}"

        # 读取原始数据
        df = (self.spark.read
            .format("kafka")
            .option("kafka.bootstrap.servers", "localhost:9092")
            .option("subscribe", source_system)
            .option("startingOffsets", "latest")
            .load())

        # 写入Bronze层
        (df.write
            .format("delta")
            .mode("append")
            .option("mergeSchema", "true")
            .partitionBy("year", "month", "day")
            .save(bronze_path))

        return bronze_path

    def build_silver_layer(self, source_system: str):
        """
        Silver层：清洗和标准化层
        - 数据清洗
        - Schema标准化
        - 数据质量检查
        - 去重和关联
        """
        bronze_path = f"{self.base_path}/bronze/{source_system}"
        silver_path = f"{self.base_path}/silver/{source_system}"

        # 读取Bronze层数据
        bronze_df = (self.spark.read
            .format("delta")
            .load(bronze_path))

        # 数据清洗和转换
        silver_df = (bronze_df
            .filter(col("value").isNotNull())
            .withColumn("data", from_json(col("value"), schema_string))
            .select(
                col("timestamp").cast("timestamp"),
                col("data.*")
            )
            .withWatermark("timestamp", "10 minutes")
            .dropDuplicates(["id", "timestamp"]))

        # 数据质量检查
        quality_checks = self.run_data_quality_checks(silver_df)

        if quality_checks["passed_rate"] < 0.95:
            raise Exception("Data quality check failed")

        # 写入Silver层（支持UPSERT）
        from delta.tables import DeltaTable

        if DeltaTable.isDeltaTable(self.spark, silver_path):
            # MERGE操作（CDC支持）
            (DeltaTable.forPath(self.spark, silver_path)
                .alias("target")
                .merge(
                    silver_df.alias("source"),
                    "target.id = source.id AND target.timestamp < source.timestamp"
                )
                .whenMatchedUpdateAll()
                .whenNotMatchedInsertAll()
                .execute())
        else:
            # 首次写入
            (silver_df.write
                .format("delta")
                .mode("overwrite")
                .partitionBy("year", "month", "day")
                .save(silver_path))

        return silver_path

    def build_gold_layer(self, business_domain: str):
        """
        Gold层：业务聚合层
        - 星型模型
        - 聚合指标
        - 优化查询性能
        - 直接服务于BI和ML
        """
        silver_path = f"{self.base_path}/silver/{business_domain}"
        gold_path = f"{self.base_path}/gold/{business_domain}"

        # 读取Silver层数据
        silver_dfs = {}
        for entity in ["users", "orders", "products"]:
            path = f"{silver_path}/{entity}"
            silver_dfs[entity] = (self.spark.read
                .format("delta")
                .load(path))

        # 构建星型模型
        # 事实表
        fact_orders = (silver_dfs["orders"]
            .groupBy("date", "product_id", "user_id")
            .agg(
                sum("amount").alias("total_amount"),
                count("*").alias("order_count"),
                avg("amount").alias("avg_order_value")
            ))

        # 维度表
        dim_products = silver_dfs["products"].select(
            "product_id", "name", "category", "brand"
        ).distinct()

        dim_users = silver_dfs["users"].select(
            "user_id", "name", "email", "segment"
        ).distinct()

        # 写入Gold层
        (fact_orders.write
            .format("delta")
            .mode("overwrite")
            .partitionBy("date")
            .save(f"{gold_path}/fact_orders"))

        (dim_products.write
            .format("delta")
            .mode("overwrite")
            .save(f"{gold_path}/dim_products"))

        (dim_users.write
            .format("delta")
            .mode("overwrite")
            .save(f"{gold_path}/dim_users"))

        # 创建优化视图
        self.create_optimized_views(gold_path)

        return gold_path

    def run_data_quality_checks(self, df):
        """数据质量检查"""
        total = df.count()
        not_null = df.filter(col("id").isNotNull()).count()

        return {
            "total_records": total,
            "not_null_records": not_null,
            "null_rate": 1 - (not_null / total),
            "passed_rate": not_null / total
        }

    def create_optimized_views(self, gold_path: str):
        """创建优化视图，加速查询"""
        # 创建Z-ORDER索引
        for table in ["fact_orders", "dim_products", "dim_users"]:
            table_path = f"{gold_path}/{table}"

            (DeltaTable.forPath(self.spark, table_path)
                .optimize()
                .executeZOrderBy("date"))

            # 优化数据文件
            (DeltaTable.forPath(self.spark, table_path)
                .optimize()
                .executeCompaction())

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120

# 表格式对比
class TableFormatComparison:
    """
    功能对比表：
    ┌──────────────┬────────────┬──────────┬──────────┐
    │    特性      │ Delta Lake │  Iceberg │   Hudi   │
    ├──────────────┼────────────┼──────────┼──────────┤
    │ ACID事务     │     ✓      │    ✓     │    ✓     │
    │ Schema演变   │     ✓      │    ✓     │    ✓     │
    │ Time Travel  │     ✓      │    ✓     │    ✓     │
    │ DELETE支持   │     ✓      │    ✓     │    ✓     │
    │ UPDATE支持   │     ✓      │    ✓     │    ✓     │
    │ Spark集成    │   优秀     │   良好    │   良好    │
    │ 社区活跃度    │   非常高   │    高     │    中     │
    │ 学习曲线      │   平缓     │   陡峭    │   陡峭    │
    └──────────────┴────────────┴──────────┴──────────┘
    """

    def delta_lake_example(self):
        """Delta Lake示例"""
        from delta import *

        # 创建表
        data = spark.range(0, 1000)
        data.write.format("delta").save("/tmp/delta-table")

        # 时间旅行查询
        df_old = (spark.read
            .format("delta")
            .option("timestampAsOf", "2024-01-01")
            .load("/tmp/delta-table"))

        # 更新操作
        from delta.tables import DeltaTable
        deltaTable = DeltaTable.forPath(spark, "/tmp/delta-table")

        (deltaTable
            .update("id % 2 == 0")
            .set("id", col("id") + 1000))

        # 删除操作
        (deltaTable
            .delete("id > 500"))

        # 合并操作（CDC）
        new_data = spark.range(1000, 1500)
        (deltaTable
            .alias("old")
            .merge(
                new_data.alias("new"),
                "old.id = new.id"
            )
            .whenMatchedUpdateAll()
            .whenNotMatchedInsertAll()
            .execute())

    def iceberg_example(self):
        """Apache Iceberg示例"""
        # 创建Iceberg表
        spark.sql("""
            CREATE TABLE hive_prod.db.orders (
                order_id BIGINT,
                user_id BIGINT,
                product_id INT,
                amount DECIMAL(10, 2),
                order_date TIMESTAMP
            )
            USING iceberg
            PARTITIONED BY (days(order_date))
            LOCATION 's3://my-bucket/warehouse/orders'
        """)

        # Schema演变
        spark.sql("""
            ALTER TABLE hive_prod.db.orders
            ADD COLUMN discount DECIMAL(5, 2)
        """)

        # 时间旅行
        spark.sql("""
            SELECT * FROM hive_prod.db.orders
            VERSION AS OF 5
        """)

        # 分区演变
        spark.sql("""
            ALTER TABLE hive_prod.db.orders
            SET PARTITION SPECIES (days(order_date))
        """)

    def hudi_example(self):
        """Apache Hudi示例"""
        # 创建Hudi表
        hudi_options = {
            'hoodie.table.name': 'orders',
            'hoodie.datasource.write.recordkey.field': 'order_id',
            'hoodie.datasource.write.partitionpath.field': 'order_date',
            'hoodie.datasource.write.precombine.field': 'updated_at',
            'hoodie.upsert.shuffle.parallelism': 200,
            'hoodie.cleaner.policy': 'KEEP_LATEST_COMMITS',
            'hoodie.cleaner.commits.retained': 10
        }

        # 写入数据
        (df.write
            .format("hudi")
            .options(**hudi_options)
            .mode("overwrite")
            .save("s3://my-bucket/hudi/orders"))

        # 增量查询
        (spark.read
            .format("hudi")
            .load("s3://my-bucket/hudi/orders")
            .createTempView("orders"))

        spark.sql("""
            SELECT * FROM orders
            WHERE `_hoodie_commit_time` > '2024-01-01'
        """)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

class DataLakeOptimizer:
    """数据湖优化器"""

    def optimize_partitioning(self):
        """
        分区策略设计原则：
        1. 分区基数 < 1000
        2. 分区大小 > 1GB
        3. 按查询频率高的字段分区
        4. 避免过细分区

        好的分区：
        - date (天级别)
        - region (地区)
        - product_category (类别)

        差的分区：
        - user_id (基数太大)
        - timestamp (过细分区)
        """

        # 推荐的分区策略
        partition_strategy = {
            "orders": ["year", "month", "day"],
            "user_events": ["date", "event_type"],
            "logs": ["date", "level", "service"]
        }

        return partition_strategy

    def optimize_file_size(self, table_path: str):
        """
        文件大小优化
        - 目标大小：128MB - 1GB
        - 太小：大量小文件，NameNode压力大
        - 太大：查询时需要读取整个文件
        """

        from delta.tables import DeltaTable

        delta_table = DeltaTable.forPath(self.spark, table_path)

        # 文件压缩
        (delta_table
            .optimize()
            .executeCompaction())

        # 数据倾斜处理
        (delta_table
            .optimize()
            .executeZOrderBy("date"))  # Z-ORDER优化

    def implement_vacuum(self, table_path: str, retention_days: int = 7):
        """
        清理旧版本数据
        - 定期执行以节省存储
        - 保留足够时间点用于回滚
        """

        from delta.tables import DeltaTable

        delta_table = DeltaTable.forPath(self.spark, table_path)

        # 保留最近7天的数据
        delta_table.vacuum(retention_hours * retention_days * 24)

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134

# 完整的ETL管道
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.spark.operators.spark_submit import SparkSubmitOperator
from datetime import datetime, timedelta

def build_data_pipeline():
    """构建数据管道"""

    default_args = {
        'owner': 'data-team',
        'depends_on_past': False,
        'start_date': datetime(2024, 1, 1),
        'email_on_failure': True,
        'email_on_retry': False,
        'retries': 2,
        'retry_delay': timedelta(minutes=5)
    }

    dag = DAG(
        'data_lake_pipeline',
        default_args=default_args,
        description='数据湖ETL管道',
        schedule_interval='0 2 * * *',  # 每天凌晨2点
        catchup=False,
        tags=['datalake', 'etl']
    )

    # 任务1：从Kafka摄取数据到Bronze层
    kafka_to_bronze = SparkSubmitOperator(
        task_id='kafka_to_bronze',
        application='jobs/ingest_kafka_to_bronze.py',
        name='ingest-kafka-to-bronze',
        conn_id='spark_default',
        conf={
            'spark.dynamicAllocation.enabled': 'true',
            'spark.dynamicAllocation.maxExecutors': '10',
            'spark.executor.memory': '4g',
            'spark.executor.cores': '2'
        },
        dag=dag
    )

    # 任务2：清洗数据到Silver层
    bronze_to_silver = SparkSubmitOperator(
        task_id='bronze_to_silver',
        application='jobs/bronze_to_silver.py',
        name='bronze-to-silver',
        conf={
            'spark.sql.adaptive.enabled': 'true',
            'spark.sql.adaptive.coalescePartitions.enabled': 'true'
        },
        dag=dag
    )

    # 任务3：构建聚合表到Gold层
    silver_to_gold = SparkSubmitOperator(
        task_id='silver_to_gold',
        application='jobs/silver_to_gold.py',
        name='silver-to-gold',
        dag=dag
    )

    # 任务4：数据质量检查
    data_quality_check = PythonOperator(
        task_id='data_quality_check',
        python_callable=run_quality_checks,
        dag=dag
    )

    # 任务5：数据血缘更新
    update_lineage = PythonOperator(
        task_id='update_lineage',
        python_callable=update_data_lineage,
        dag=dag
    )

    # 任务6：发送报告
    send_report = PythonOperator(
        task_id='send_report',
        python_callable=send_pipeline_report,
        dag=dag
    )

    # 设置任务依赖
    kafka_to_bronze >> bronze_to_silver >> silver_to_gold
    silver_to_gold >> data_quality_check >> update_lineage >> send_report

    return dag

# 主函数
def run_quality_checks(**context):
    """运行数据质量检查"""
    from quality_checker import DataQualityChecker

    checker = DataQualityChecker()

    # 检查空值率
    null_checks = checker.check_nullity("gold.fact_orders")

    # 检查数据范围
    range_checks = checker.check_ranges("gold.fact_orders", {
        "total_amount": (0, 1000000),
        "order_count": (0, 10000)
    })

    # 检查数据完整性
    integrity_checks = checker.check_referential_integrity(
        "gold.fact_orders",
        {"dim_products": ["product_id"], "dim_users": ["user_id"]}
    )

    # 生成质量报告
    report = {
        "null_checks": null_checks,
        "range_checks": range_checks,
        "integrity_checks": integrity_checks,
        "overall_score": calculate_overall_score([
            null_checks, range_checks, integrity_checks
        ])
    }

    if report["overall_score"] < 0.9:
        raise Exception(f"Data quality score too low: {report['overall_score']}")

    return report

def calculate_overall_score(checks):
    """计算总体质量分数"""
    scores = []
    for check in checks:
        if isinstance(check, dict):
            scores.append(check.get("score", 1.0))
    return sum(scores) / len(scores) if scores else 1.0

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

class StorageOptimizer:
    """存储优化策略"""

    def implement_tiered_storage(self):
        """
        分层存储策略：
        - 热数据：SSD（最近7天）
        - 温数据：HDD（30天）
        - 冷数据：S3 Glacier（归档）
        """

        # 生命周期策略
        lifecycle_policy = {
            "rules": [
                {
                    "id": "hot-to-warm",
                    "transition": {
                        "days": 7,
                        "storage_class": "STANDARD_IA"
                    }
                },
                {
                    "id": "warm-to-cold",
                    "transition": {
                        "days": 30,
                        "storage_class": "GLACIER"
                    }
                },
                {
                    "id": "expiration",
                    "expiration": {
                        "days": 365
                    }
                }
            ]
        }

        return lifecycle_policy

    def implement_columnar_format(self):
        """
        列式存储优化
        - Parquet：压缩率高，查询快
        - ORC：适合复杂类型
        - Avro：适合写密集
        """

        format_recommendations = {
            "分析查询": "parquet",
            "流式写入": "avro",
            "复杂嵌套": "orc",
            "读取密集": "parquet",
            "写入密集": "avro"
        }

        return format_recommendations