echo “=== 电商数仓分层构建 ===”
echo “Date: $(date)”

DT=$1
if [ -z “$DT” ]; then
  DT=$(date -d “-1 day” +%Y-%m-%d)
fi

echo “Processing date: $DT”

# ODS层数据同步
echo “Step 1: ODS Layer Data Sync”
hive -e “
— 同步订单数据
INSERT OVERWRITE TABLE ods_trade.ods_trade_order_di PARTITION(dt=’$DT’)
SELECT * FROM source_orders WHERE dt = ‘$DT’;

— 同步用户数据
INSERT OVERWRITE TABLE ods_trade.ods_trade_user_df PARTITION(dt=’$DT’)
SELECT * FROM source_users;
“

# DWD层数据处理
echo “Step 2: DWD Layer Data Processing”
hive -e “
INSERT OVERWRITE TABLE dwd_trade.dwd_trade_order_di PARTITION(dt=’$DT’)
SELECT
  o.order_id, o.user_id, u.user_name, u.user_level,
  o.product_id, p.product_name, p.category_id, c.category_name,
  o.order_amount, o.order_status, pay.payment_amount, pay.payment_time,
  o.create_time, CURRENT_TIMESTAMP, ‘$DT’
FROM ods_trade.ods_trade_order_di o
LEFT JOIN dim_trade.dim_user_df u ON o.user_id = u.user_id
LEFT JOIN dim_trade.dim_product_df p ON o.product_id = p.product_id
LEFT JOIN dim_trade.dim_category_df c ON p.category_id = c.category_id
LEFT JOIN ods_trade.ods_trade_payment_di pay ON o.order_id = pay.order_id AND pay.dt = ‘$DT’
WHERE o.dt = ‘$DT’;
“

# DWS层数据汇总
echo “Step 3: DWS Layer Data Aggregation”
hive -e “
INSERT OVERWRITE TABLE dws_trade.dws_trade_user_order_1d PARTITION(dt=’$DT’)
SELECT
  user_id, user_name, user_level,
  COUNT(*) AS order_count,
  SUM(order_amount) AS order_amount,
  SUM(payment_amount) AS payment_amount,
  MIN(create_time) AS first_order_time,
  MAX(create_time) AS last_order_time,
  CURRENT_TIMESTAMP, ‘$DT’
FROM dwd_trade.dwd_trade_order_di
WHERE dt = ‘$DT’
GROUP BY user_id, user_name, user_level;
“

echo “=== 构建完成 ===”