数据工程实战项目¶

# 业务需求
requirements = {
    'sales_analysis': {
        'description': '销售数据分析',
        'metrics': ['total_sales', 'average_order_value', 'conversion_rate'],
        'dimensions': ['product', 'category', 'customer', 'time']
    },
    'customer_analysis': {
        'description': '客户分析',
        'metrics': ['customer_lifetime_value', 'retention_rate', 'churn_rate'],
        'dimensions': ['customer_segment', 'acquisition_channel', 'time']
    },
    'inventory_analysis': {
        'description': '库存分析',
        'metrics': ['inventory_turnover', 'stockout_rate', 'reorder_point'],
        'dimensions': ['product', 'warehouse', 'time']
    }
}

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'data_team',
    'start_date': datetime(2024, 1, 1),
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'ecommerce_warehouse',
    default_args=default_args,
    description='Ecommerce data warehouse ETL',
    schedule='@daily',  # 使用新参数名（Airflow 2.x）
    catchup=False,
)

def extract_sales():
    """抽取销售数据"""
    print("Extracting sales data...")
    # 实际的数据抽取逻辑
    return 'sales_data.csv'

def transform_sales():
    """转换销售数据"""
    print("Transforming sales data...")
    # 实际的数据转换逻辑
    return 'transformed_sales.csv'

def load_sales():
    """加载销售数据"""
    print("Loading sales data...")
    # 实际的数据加载逻辑
    return 'load_complete'

extract_task = PythonOperator(
    task_id='extract',
    python_callable=extract_sales,
    dag=dag
)

transform_task = PythonOperator(
    task_id='transform',
    python_callable=transform_sales,
    dag=dag
)

load_task = PythonOperator(
    task_id='load',
    python_callable=load_sales,
    dag=dag
)

extract_task >> transform_task >> load_task

from kafka import KafkaProducer
import json

producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

# 发送实时数据
producer.send('sales', value={
        'product_id': 1,
        'quantity': 10,
        'amount': 100.0,
        'timestamp': datetime.now().isoformat()
    })
producer.flush()

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment

env = StreamExecutionEnvironment.get_execution_environment()
t_env = StreamTableEnvironment.create(env)

t_env.execute_sql("""
    CREATE TABLE sales (
        product_id INT,
        quantity INT,
        amount DECIMAL(10,2),
        timestamp TIMESTAMP(3),
        WATERMARK FOR timestamp AS timestamp - INTERVAL '5' SECOND
    ) WITH (
        'connector' = 'kafka',
        'topic' = 'sales',
        'properties.bootstrap.servers' = 'localhost:9092',
        'format' = 'json'
    )
""")

result = t_env.sql_query("""
    SELECT
        product_id,
        COUNT(*) as order_count,
        SUM(quantity) as total_quantity,
        SUM(amount) as total_amount
    FROM sales
    GROUP BY
        product_id,
        TUMBLE(timestamp, INTERVAL '1' MINUTE)
""")

result.execute().print()

import great_expectations as gx
import pandas as pd

# 创建 Data Context
context = gx.get_context()

# 读取数据并创建 Data Source
df = pd.read_csv('sales.csv')
data_source = context.data_sources.add_pandas(name="sales_source")
data_asset = data_source.add_dataframe_asset(name="sales_asset")
batch_definition = data_asset.add_batch_definition_whole_dataframe("sales_batch")

# 创建 Expectation Suite
suite = context.suites.add(
    gx.ExpectationSuite(name="sales_suite")
)

# 添加 Expectations
suite.add_expectation(
    gx.expectations.ExpectColumnToExist(column="product_id")
)

suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeBetween(
        column='quantity',
        min_value=1,
        max_value=1000
    )
)

suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeBetween(
        column='amount',
        min_value=0,
        max_value=100000
    )
)

# 创建 Validation Definition 并验证
validation_definition = context.validation_definitions.add(
    gx.ValidationDefinition(
        name="sales_validation",
        data=batch_definition,
        suite=suite
    )
)
result = validation_definition.run(batch_parameters={"dataframe": df})

# 输出结果
print(f"Validation success: {result.success}")
for exp_result in result.results:
    print(f"  {exp_result.expectation_config.type}: {exp_result.success}")

# 导入 DataHub REST 发射器，用于向 DataHub 服务端推送元数据
from datahub.emitter.rest_emitter import DatahubRestEmitter
# 导入元数据变更提案包装器，封装元数据变更请求
from datahub.emitter.mcp import MetadataChangeProposalWrapper
# 导入各类元数据 Schema 定义类
from datahub.metadata.schema_classes import (
    DatasetPropertiesClass,       # 数据集属性（名称、描述等）
    UpstreamLineageClass,         # 上游血缘关系
    UpstreamClass,                # 上游数据集
    DatasetLineageTypeClass,      # 血缘类型（转换、复制等）
)

# 创建 DataHub Emitter
emitter = DatahubRestEmitter(
    gms_server="http://localhost:8080"
)

# 定义数据集 URN
dataset_urn = "urn:li:dataset:(urn:li:dataPlatform:hive,sales_data,PROD)"

# 发送数据集属性
dataset_properties = MetadataChangeProposalWrapper(
    entityUrn=dataset_urn,
    aspect=DatasetPropertiesClass(
        name="Sales Data",
        description="Daily sales data",
        customProperties={"team": "data_engineering"}
    )
)
emitter.emit(dataset_properties)

# 发送血缘关系（上游依赖），描述数据集之间的转换关系
upstream_urn = "urn:li:dataset:(urn:li:dataPlatform:mysql,raw_orders,PROD)"
lineage = MetadataChangeProposalWrapper(
    entityUrn=dataset_urn,
    aspect=UpstreamLineageClass(
        upstreams=[
            UpstreamClass(
                dataset=upstream_urn,
                type=DatasetLineageTypeClass.TRANSFORMED
            )
        ]
    )
)
emitter.emit(lineage)

print(f"DataHub metadata emitted for: {dataset_urn}")

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment

env = StreamExecutionEnvironment.get_execution_environment()
t_env = StreamTableEnvironment.create(env)

t_env.execute_sql("""
    CREATE TABLE user_events (
        user_id BIGINT,
        product_id BIGINT,
        event_type STRING,
        timestamp TIMESTAMP(3)
    ) WITH (
        'connector' = 'kafka',
        'topic' = 'user_events',
        'properties.bootstrap.servers' = 'localhost:9092',
        'format' = 'json'
    )
""")

# 计算用户特征
user_features = t_env.sql_query("""
    SELECT
        user_id,
        COUNT(*) as event_count,
        COUNT(DISTINCT product_id) as product_diversity,
        AVG(CASE WHEN event_type = 'view' THEN 1 ELSE 0 END) as view_rate,
        AVG(CASE WHEN event_type = 'purchase' THEN 1 ELSE 0 END) as purchase_rate
    FROM user_events
    GROUP BY user_id,
    TUMBLE(timestamp, INTERVAL '1' HOUR)
""")

user_features.execute().print()