数据工程师快速上手

-- 在 Notebook 中新建 SQL 单元格（单元格左上角语言选择 SQL），先执行以下建表语句

-- 1. 创建 Schema（如果不存在）
CREATE SCHEMA IF NOT EXISTS `DataLakeCatalog`.`wedata_demo_db`;

-- 2. 创建表（如果不存在）
CREATE TABLE IF NOT EXISTS `DataLakeCatalog`.`wedata_demo_db`.`quickstart_amazon_orders` (
    `order_id`        STRING COMMENT '订单号',
    `user_id`         STRING COMMENT '用户ID',
    `product_id`      STRING COMMENT '商品ID',
    `product_name`    STRING COMMENT '商品名',
    `category`        STRING COMMENT '商品品类',
    `unit_price`      DECIMAL(10,2) COMMENT '单价',
    `quantity`        INT COMMENT '数量',
    `total_amount`    DECIMAL(12,2) COMMENT '订单金额',
    `order_status`    STRING COMMENT '订单状态',
    `payment_method`  STRING COMMENT '支付方式',
    `order_date`      STRING COMMENT '下单日期',
    `state_code`      STRING COMMENT '州简码',
    `state`           STRING COMMENT '州名',
    `city`            STRING COMMENT '城市'
) COMMENT '样例订单表';

-- 3. 插入样例数据
INSERT INTO `DataLakeCatalog`.`wedata_demo_db`.`quickstart_amazon_orders`
(`order_id`, `user_id`, `product_id`, `product_name`, `category`, `unit_price`, `quantity`, `total_amount`, `order_status`, `payment_method`, `order_date`, `state_code`, `state`, `city`)
VALUES
('ORD-001','U001','P001','Towel Set','Home',CAST(15.99 AS DECIMAL(10,2)),2,CAST(31.98 AS DECIMAL(12,2)),'Delivered','CreditCard','01/03/2024','CA','California','Los Angeles'),
('ORD-002','U002','P002','Bedsheet','Home',CAST(25.50 AS DECIMAL(10,2)),1,CAST(25.50 AS DECIMAL(12,2)),'Shipped','PayPal','02/03/2024','NY','New York','New York City'),
('ORD-003','U003','P003','Curtain','Home',CAST(40.00 AS DECIMAL(10,2)),3,CAST(120.00 AS DECIMAL(12,2)),'Delivered','CreditCard','03/03/2024','TX','Texas','Houston'),
('ORD-004','U004','P004','Carpet','Home',CAST(120.50 AS DECIMAL(10,2)),1,CAST(120.50 AS DECIMAL(12,2)),'Cancelled','ApplePay','04/03/2024','FL','Florida','Miami'),
('ORD-005','U005','P005','Pillow','Home',CAST(12.99 AS DECIMAL(10,2)),4,CAST(51.96 AS DECIMAL(12,2)),'Delivered','CreditCard','05/03/2024','WA','Washington','Seattle'),
('ORD-006','U006','P006','Mattress','Home',CAST(299.99 AS DECIMAL(10,2)),1,CAST(299.99 AS DECIMAL(12,2)),'Delivered','DebitCard','06/03/2024','IL','Illinois','Chicago'),
('ORD-007','U007','P007','Blanket','Home',CAST(35.00 AS DECIMAL(10,2)),2,CAST(70.00 AS DECIMAL(12,2)),'Shipped','CreditCard','07/03/2024','AZ','Arizona','Phoenix'),
('ORD-008','U001','P001','Towel Set','Home',CAST(15.99 AS DECIMAL(10,2)),5,CAST(79.95 AS DECIMAL(12,2)),'Delivered','CreditCard','08/03/2024','CA','California','San Diego'),
('ORD-009','U008','P008','Rug','Home',CAST(65.00 AS DECIMAL(10,2)),1,CAST(65.00 AS DECIMAL(12,2)),'Delivered','PayPal','09/03/2024','NV','Nevada','Las Vegas'),
('ORD-010','U009','P009','Comforter','Home',CAST(85.00 AS DECIMAL(10,2)),1,CAST(85.00 AS DECIMAL(12,2)),'Shipped','CreditCard','10/03/2024','OR','Oregon','Portland');

-- 4. 验证数据已插入
SELECT * FROM `DataLakeCatalog`.`wedata_demo_db`.`quickstart_amazon_orders`;

df_orders = spark.sql("""
    SELECT *
    FROM `DataLakeCatalog`.`wedata_demo_db`.`quickstart_amazon_orders`
""")
display(df_orders)
print(f"总行数: {df_orders.count()}")

from pyspark.sql import functions as F

# 过滤已取消订单
df_valid = df_orders.filter(F.col("order_status") != "Cancelled")

# 把 dd/MM/yyyy 文本日期解析为 date，并衍生年月
df_valid = (
    df_valid
    .withColumn("order_date_parsed", F.to_date("order_date", "dd/MM/yyyy"))
    .withColumn("order_year", F.year("order_date_parsed"))
    .withColumn("order_month", F.month("order_date_parsed"))
)

display(df_valid.select("order_id", "order_date", "order_date_parsed", "order_year", "order_month"))

df_state_revenue = (
    df_valid.groupBy("state", "state_code")
    .agg(
        F.count("order_id").alias("order_count"),
        F.sum("total_amount").alias("total_revenue"),
        F.avg("total_amount").alias("avg_order_value"),
        F.countDistinct("user_id").alias("unique_customers"),
    )
    .orderBy(F.desc("total_revenue"))
)

display(df_state_revenue)

spark.sql("""
    CREATE TABLE IF NOT EXISTS `DataLakeCatalog`.`wedata_demo_db`.`quickstart_state_revenue` (
        state            STRING  COMMENT '州名',
        state_code       STRING  COMMENT '州简码',
        order_count      BIGINT  COMMENT '订单数',
        total_revenue    DOUBLE  COMMENT '总收入',
        avg_order_value  DOUBLE  COMMENT '客单价',
        unique_customers BIGINT  COMMENT '去重用户数'
    ) COMMENT '按州维度的收入汇总'
""")

(
    df_state_revenue
    .select("state", "state_code", "order_count", "total_revenue", "avg_order_value", "unique_customers")
    .write.mode("overwrite")
    .insertInto("`DataLakeCatalog`.`wedata_demo_db`.`quickstart_state_revenue`")
)

display(spark.sql("SELECT * FROM `DataLakeCatalog`.`wedata_demo_db`.`quickstart_state_revenue`"))

sync_mysql2dlc  ───────▶  pyspark_transform
(离线数据接入)            (Notebook)