#海量数据处理

-- ODS 层建表：用户行为日志原始表
CREATE TABLE ods.user_behavior_log (
    -- 原始字段，直接从 Kafka topic 的 Avro schema 映射
    log_id     STRING,
    event_type STRING,        -- click, view, add_cart, purchase
    user_id    STRING,
    device_id  STRING,
    ip         STRING,
    url        STRING,
    referrer   STRING,
    user_agent STRING,
    timestamp  BIGINT,         -- 毫秒时间戳
    params     MAP<STRING, STRING>,  -- 动态扩展字段
    -- 审计字段
    dt         STRING,         -- 分区字段：2024-01-15
    hp_time    TIMESTAMP      -- 数据进入数仓的时间
) PARTITIONED BY (dt STRING)
STORED AS PARQUET
TBLPROPERTIES ('parquet.compression' = 'SNAPPY');

-- DWD 层建表：用户行为明细表（已清洗）
CREATE TABLE dwd.user_behavior_detail (
    -- 核心维度
    user_id        STRING,
    user_type      STRING,      -- new / old
    user_level     STRING,      -- v1-v8 会员等级
    province       STRING,      -- 用户归属省份
    city           STRING,      -- 用户归属城市

    -- 行为事实
    event_type     STRING,      -- click / view / add_cart / purchase / refund
    product_id     STRING,      -- 商品 ID（非购买事件为空）
    product_name   STRING,
    category_id    STRING,
    category_name  STRING,
    brand_id       STRING,
    brand_name     STRING,
    shop_id        STRING,
    shop_name      STRING,
    price          DECIMAL(12,2),  -- 商品价格（购买事件）
    quantity       INT,          -- 购买数量（购买事件）

    -- 时间维度
    event_time     TIMESTAMP,   -- 事件发生时间
    event_date     STRING,      -- 事件日期：2024-01-15

    -- 地域维度（从用户维表关联）
    ip             STRING,
    ip_province    STRING,
    ip_city        STRING,

    -- 来源维度
    source         STRING,      -- app / h5 / pc / mini_program
    channel        STRING,      -- 应用市场 / 分享链接 / 搜索
    utm_source     STRING,
    utm_medium     STRING,
    utm_campaign   STRING
) PARTITIONED BY (dt STRING)
STORED AS PARQUET;

-- DWD 层数据清洗任务（每日凌晨调度）
INSERT OVERWRITE TABLE dwd.user_behavior_detail PARTITION (dt = '${bizdate}')
SELECT
    -- 基础信息
    user_id,
    CASE WHEN u.first_order_time IS NOT NULL THEN 'old' ELSE 'new' END as user_type,
    COALESCE(u.level, 'v1') as user_level,
    COALESCE(u.province, '未知') as province,
    COALESCE(u.city, '未知') as city,

    -- 行为类型标准化
    CASE
        WHEN event_type = 'product_click' THEN 'click'
        WHEN event_type = 'sku_view' THEN 'view'
        WHEN event_type = 'add_to_cart' THEN 'add_cart'
        WHEN event_type = 'order_submit' THEN 'purchase'
        WHEN event_type = 'order_refund' THEN 'refund'
        ELSE event_type
    END as event_type,

    -- 商品信息
    product_id,
    product_name,
    category_id,
    category_name,
    brand_id,
    brand_name,
    shop_id,
    shop_name,
    CAST(price AS DECIMAL(12,2)) as price,
    CAST(quantity AS INT) as quantity,

    -- 时间和地域
    FROM_UNIXTIME(timestamp / 1000) as event_time,
    '${bizdate}' as event_date,
    ip,
    ip_to_province(ip) as ip_province,
    ip_to_city(ip) as ip_city,

    -- 来源信息
    source,
    channel,
    utm_source,
    utm_medium,
    utm_campaign

FROM ods.user_behavior_log
WHERE dt = '${bizdate}'
  -- 过滤爬虫流量
  AND NOT is_crawler(user_agent, ip)
  -- 过滤测试用户
  AND user_id NOT IN (${test_user_ids})
  -- 过滤内部流量
  AND ip NOT IN (${internal_ip_ranges})
  -- 过滤异常数据
  AND user_id IS NOT NULL
  AND LENGTH(user_id) > 0;

-- DWS 层建表：用户粒度日活汇总
CREATE TABLE dws.user_daily_summary (
    user_id         STRING,
    dt              STRING,        -- 日期

    -- 行为汇总
    pv              BIGINT,        -- 页面浏览次数
    click_cnt       BIGINT,        -- 点击次数
    cart_cnt        BIGINT,        -- 加购次数
    order_cnt       BIGINT,        -- 下单次数
    order_amount    DECIMAL(14,2), -- 下单金额
    pay_cnt         BIGINT,        -- 支付次数
    pay_amount      DECIMAL(14,2), -- 支付金额

    -- 商品维度
    product_cnt     BIGINT,        -- 访问商品数
    category_cnt    BIGINT,        -- 访问类目数

    -- 时间特征
    first_event_time TIMESTAMP,    -- 当天首次行为时间
    last_event_time  TIMESTAMP,   -- 当天末次行为时间
    duration_seconds  BIGINT       -- 活跃时长（秒）
) PARTITIONED BY (dt STRING)
STORED AS PARQUET;

-- DWS 层计算任务
INSERT OVERWRITE TABLE dws.user_daily_summary PARTITION (dt = '${bizdate}')
SELECT
    user_id,
    '${bizdate}' as dt,

    COUNT(1) as pv,
    SUM(CASE WHEN event_type = 'click' THEN 1 ELSE 0 END) as click_cnt,
    SUM(CASE WHEN event_type = 'add_cart' THEN 1 ELSE 0 END) as cart_cnt,
    SUM(CASE WHEN event_type = 'purchase' THEN 1 ELSE 0 END) as order_cnt,
    SUM(CASE WHEN event_type = 'purchase' THEN price ELSE 0 END) as order_amount,
    SUM(CASE WHEN is_paid THEN 1 ELSE 0 END) as pay_cnt,
    SUM(CASE WHEN is_paid THEN price ELSE 0 END) as pay_amount,

    COUNT(DISTINCT product_id) as product_cnt,
    COUNT(DISTINCT category_id) as category_cnt,

    MIN(event_time) as first_event_time,
    MAX(event_time) as last_event_time,
    UNIX_TIMESTAMP(MAX(event_time)) - UNIX_TIMESTAMP(MIN(event_time)) as duration_seconds

FROM dwd.user_behavior_detail
WHERE dt = '${bizdate}'
  AND user_id IS NOT NULL
GROUP BY user_id;

-- ADS 层：运营大盘实时数据（供大屏使用）
CREATE TABLE ads.operation_dashboard (
    dt              STRING,
    -- 核心指标
    dau             BIGINT,       -- 日活用户数
    new_user_cnt    BIGINT,       -- 新增用户数
    order_cnt       BIGINT,       -- 订单数
    order_amount    DECIMAL(16,2),-- GMV
    pay_rate        DECIMAL(6,4), -- 支付转化率
    avg_order_price DECIMAL(10,2),-- 客单价

    -- 商品维度
    top_category    STRING,       -- GMV 最高的类目
    top_product_id  STRING,       -- GMV 最高的商品

    -- 时间戳
    update_time     TIMESTAMP
) STORED AS PARQUET
TBLPROPERTIES ('parquet.compression' = 'SNAPPY');

-- ClickHouse 建表：用户行为分析表
CREATE TABLE ch.user_behaviors (
    user_id       UInt32,
    event_type    Enum8('click'=1, 'view'=2, 'cart'=3, 'purchase'=4, 'refund'=5),
    product_id    UInt32,
    category_id   UInt32,
    brand_id      UInt32,
    price         Decimal(10,2),
    quantity       UInt16,
    province      FixedString(24),
    city          FixedString(24),
    source        UInt8,     -- 1=app 2=h5 3=pc 4=小程序
    channel       String,
    event_time    DateTime,
    dt            Date
) ENGINE = MergeTree()
PARTITION BY dt
ORDER BY (dt, event_type, user_id)  -- 排序键：决定数据如何物理排列
SETTINGS index_granularity = 8192;  -- 索引粒度：每隔 8192 行建一个稀疏索引

-- 查询示例：计算各渠道的 GMV（ClickHouse 秒级返回上亿行数据）
SELECT
    channel,
    count() as cnt,
    sum(price * quantity) as gmv
FROM user_behaviors
WHERE dt >= '2024-01-01' AND dt <= '2024-01-15'
  AND event_type = 'purchase'
GROUP BY channel
ORDER BY gmv DESC
LIMIT 100;

-- ClickHouse 冷热分层存储（需要 ClickHouse Cloud 或自建配置）
-- 语法示例（ClickHouse 22.8+）
CREATE TABLE user_behaviors (
    user_id     UInt32,
    event_type  String,
    product_id  UInt32,
    event_time  DateTime,
    dt          Date
) ENGINE = MergeTree()
ORDER BY (dt, event_type, user_id)
TTL dt + INTERVAL 30 DAY  -- 30 天后自动移动到冷存储
SETTINGS storage_policy = 'tiered';  -- tiered: 热 + 冷分层

-- 查看存储使用情况
SELECT
    table,
    sum(rows) as total_rows,
    sum(bytes_on_disk) as total_bytes,
    sum(bytes_on_disk) / 1024 / 1024 / 1024 as size_gb
FROM system.parts
WHERE table LIKE 'dwd.%'
GROUP BY table
ORDER BY size_gb DESC;

-- 错误的写法：直接 JOIN
SELECT a.user_id, b.user_name, c.order_cnt
FROM dwd.user a
JOIN dim.user_info b ON a.user_id = b.user_id
JOIN ads.user_orders c ON a.user_id = c.user_id  -- [!code error]

-- 正确的写法：子查询 + GLOBAL IN
SELECT user_id, user_name, order_cnt
FROM dwd.user a
ANY LEFT JOIN (
    SELECT user_id, user_name FROM dim.user_info
) b GLOBAL IN (SELECT user_id, user_name FROM dim.user_info)  
ON a.user_id = b.user_id
ANY LEFT JOIN (
    SELECT user_id, any(order_cnt) as order_cnt
    FROM ads.user_orders GROUP BY user_id
) c ON a.user_id = c.user_id;