LabEx
免费加入

将 Spark 订单与商品及客户数据关联

大数据Beginner
立即练习

介绍

订单知道客户和商品标识符，独立参考表则描述客户居住地区以及商品所属分类。关联（join）通过这些键连接记录，让报表能够回答按地区或商品分类统计收入等问题。

你将逐个使用对照表补充已付款订单，选择无歧义的输出列，并核对总计。还将比较客户缺失时 INNER JOIN 与 LEFT JOIN 的行为。提供的客户键和商品键都唯一；每个能匹配的订单在各对照表中应找到一条记录。重复对照键可能使关联行数成倍增加并夸大收入，因此预期关系很重要。

本独立实验准备了干净订单、两个客户、三个商品，以及单独的缺失匹配示例。你应该了解有类型读取、筛选、十进制金额、聚合和最小 Parquet 输出。无需 Hive、SQL、Hadoop、pandas 或集群配置。请使用终端。

通过明确键添加客户地区

在这一步中，你将把四笔已付款订单关联到各自唯一的客户记录。查看源关系：

cd /home/labex/project

cat /home/labex/project/data/orders.csv

cat /home/labex/project/data/customers.csv

多个订单可以共享 customer_id，而对照表中 c1 和 c2 各有一行。相等标识符连接记录；按行位置匹配不是有效规则。c1 提供 East（东区），c2 提供 West（西区）。

DataFrame join 接受另一个 DataFrame、匹配条件和关联类型。inner 只保留匹配记录。DataFrame 的 alias("o") 和 alias("c") 为输入设置简短引用名。col("o.customer_id") 标识订单侧列，col("c.customer_id") 标识对照表侧列。

两个输入都有 customer_id 列。明确的 select 选择订单侧键和所需各字段，避免重复列名引起歧义。DataFrame 别名命名输入引用，列别名命名结果字段。保存并运行完整版本，其中读取器和金额计算重复之前技能：

cat > /home/labex/project/enrich_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum as spark_sum
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType

spark = SparkSession.builder.appName("EnrichOrders").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
order_schema = StructType([
    StructField("order_id", StringType(), True),
    StructField("customer_id", StringType(), True),
    StructField("product_id", StringType(), True),
    StructField("quantity", IntegerType(), True),
    StructField("unit_price", DecimalType(10, 2), True),
    StructField("order_status", StringType(), True),
    StructField("order_date", DateType(), True),
])
orders = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")
paid = orders.filter(col("order_status") == "paid").withColumn("order_amount", col("quantity") * col("unit_price"))
customers = spark.read.option("header", True).schema("customer_id string, region string").csv("/home/labex/project/data/customers.csv")
products = spark.read.option("header", True).schema("product_id string, category string").csv("/home/labex/project/data/products.csv")

with_customers = (
    paid.alias("o")
    .join(customers.alias("c"), col("o.customer_id") == col("c.customer_id"), "inner")
    .select(
        col("o.order_id").alias("order_id"), col("o.customer_id").alias("customer_id"),
        col("o.product_id").alias("product_id"), col("o.quantity").alias("quantity"),
        col("o.unit_price").alias("unit_price"), col("o.order_status").alias("order_status"),
        col("o.order_date").alias("order_date"), col("o.order_amount").alias("order_amount"),
        col("c.region").alias("region"),
    )
)
with_customers.orderBy("order_id").show(truncate=False)
with_customers.write.mode("overwrite").parquet("/home/labex/project/output/orders-with-regions")
spark.stop()
PYTHON

bigdata-python /home/labex/project/enrich_orders.py > /home/labex/project/output/customer-join.txt

cat /home/labex/project/output/customer-join.txt

预期为四条补充地区后的已付款行。o1、o4 应为 East，o2、o6 为 West；金额仍为 20.00、25.00、50.00、30.00。四个已付款输入具有完整且唯一匹配，因此生成四个关联行。Parquet 输出保留有类型字段并添加 region，没有改变原始 CSV。

添加商品分类并核对收入

在这一步中，你将使用 product_id 添加第二个对照表。查看商品映射：

cat /home/labex/project/data/products.csv

p1 和 p2 属于 Office（办公），p3 属于 Home（家居）。按 customer_id 关联这张表会使用错误关系。保留中间的已补充订单字段，只从商品对照表选择 category。

这个完整版本保留第一次关联，再增加第二次。明确投影只保留一个 product_id，避免留下两个同名键。然后用熟悉的 groupBy 和聚合 sum 计算分类收入：

cat > /home/labex/project/enrich_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum as spark_sum
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType

spark = SparkSession.builder.appName("EnrichOrders").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
order_schema = StructType([
    StructField("order_id", StringType(), True),
    StructField("customer_id", StringType(), True),
    StructField("product_id", StringType(), True),
    StructField("quantity", IntegerType(), True),
    StructField("unit_price", DecimalType(10, 2), True),
    StructField("order_status", StringType(), True),
    StructField("order_date", DateType(), True),
])
orders = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")
paid = orders.filter(col("order_status") == "paid").withColumn("order_amount", col("quantity") * col("unit_price"))
customers = spark.read.option("header", True).schema("customer_id string, region string").csv("/home/labex/project/data/customers.csv")
products = spark.read.option("header", True).schema("product_id string, category string").csv("/home/labex/project/data/products.csv")

with_customers = (
    paid.alias("o")
    .join(customers.alias("c"), col("o.customer_id") == col("c.customer_id"), "inner")
    .select(
        col("o.order_id").alias("order_id"), col("o.customer_id").alias("customer_id"),
        col("o.product_id").alias("product_id"), col("o.quantity").alias("quantity"),
        col("o.unit_price").alias("unit_price"), col("o.order_status").alias("order_status"),
        col("o.order_date").alias("order_date"), col("o.order_amount").alias("order_amount"),
        col("c.region").alias("region"),
    )
)
enriched = (
    with_customers.alias("e")
    .join(products.alias("p"), col("e.product_id") == col("p.product_id"), "inner")
    .select(
        col("e.order_id").alias("order_id"), col("e.customer_id").alias("customer_id"),
        col("e.product_id").alias("product_id"), col("e.quantity").alias("quantity"),
        col("e.unit_price").alias("unit_price"), col("e.order_status").alias("order_status"),
        col("e.order_date").alias("order_date"), col("e.order_amount").alias("order_amount"),
        col("e.region").alias("region"), col("p.category").alias("category"),
    )
)
category_sales = enriched.groupBy("category").agg(spark_sum("order_amount").alias("revenue"))
enriched.write.mode("overwrite").parquet("/home/labex/project/output/enriched-orders")
category_sales.write.mode("overwrite").parquet("/home/labex/project/output/category-sales")
enriched.orderBy("order_id").show(truncate=False)
category_sales.orderBy("category").show(truncate=False)
spark.stop()
PYTHON

bigdata-python /home/labex/project/enrich_orders.py > /home/labex/project/output/product-join.txt

cat /home/labex/project/output/product-join.txt

补充后的结果应仍有四笔已付款订单，现在同时包含地区和分类。分类报表应显示 Home 30.00、Office 95.00，总计仍为 125.00 已付款收入。

行数保持四和收入对账一致，是两项实用检查。仅仅添加描述列，不应让二者增加。本样本的键唯一且完整；下一步单独探索缺失匹配，不改变主销售总计。

比较客户缺失时的 INNER 与 LEFT JOIN

在这一步中，你将观察对照表无法匹配时各类关联的行为。查看独立示例：

cat /home/labex/project/data/example-orders.csv

demo1 引用 c1，demo2 引用 customers 中不存在的 c_missing。先预测 INNER JOIN：应该只保留 demo1。LEFT JOIN 保留每条左侧示例记录，并为原本应由缺失右侧记录提供的字段填入 NULL。

保存独立的小程序，保持主订单补充脚本及数据不变：

cat > /home/labex/project/compare_joins.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col

spark = SparkSession.builder.appName("CompareCustomerMatches").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
examples = spark.read.option("header", True).schema("order_id string, customer_id string").csv("/home/labex/project/data/example-orders.csv")
customers = spark.read.option("header", True).schema("customer_id string, region string").csv("/home/labex/project/data/customers.csv")
inner = examples.alias("o").join(customers.alias("c"), col("o.customer_id") == col("c.customer_id"), "inner").select(col("o.order_id").alias("order_id"), col("o.customer_id").alias("customer_id"), col("c.region").alias("region"))
left = examples.alias("o").join(customers.alias("c"), col("o.customer_id") == col("c.customer_id"), "left").select(col("o.order_id").alias("order_id"), col("o.customer_id").alias("customer_id"), col("c.region").alias("region"))
print("Inner matches:")
inner.orderBy("order_id").show(truncate=False)
print("Left matches:")
left.orderBy("order_id").show(truncate=False)
inner.write.mode("overwrite").parquet("/home/labex/project/output/example-inner")
left.write.mode("overwrite").parquet("/home/labex/project/output/example-left")
spark.stop()
PYTHON

bigdata-python /home/labex/project/compare_joins.py > /home/labex/project/output/join-comparison.txt

cat /home/labex/project/output/join-comparison.txt

Inner matches 应包含 demo1/c1/East。Left matches 应包含该行，以及 demo2/c_missing/NULL。右侧地区缺失时，左侧 customer_id 仍然可用。NULL 表示没有匹配地区，不是字面名称为 NULL 的地区。

对于已确认匹配完整的报表，INNER JOIN 可能合适。LEFT JOIN 可以为质量检查或明确业务策略保留未解决记录。选择关联类型会改变哪些记录保留下来，不只是显示偏好。

使用补充字段进行地区比较

在这一步中，你将对主数据集应用已学的筛选和分组。全部分类的地区收入应为 East 70.00、West 55.00。预测只包含 Office 时的结果：East 保持 70.00，West 变为 25.00，因为它的 30.00 Home 订单被排除。

运行完整主脚本，增加两个派生报表。关联保持不变，分类条件只改变进入第二个聚合的订单行：

cat > /home/labex/project/enrich_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum as spark_sum
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType

spark = SparkSession.builder.appName("EnrichOrders").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
order_schema = StructType([
    StructField("order_id", StringType(), True),
    StructField("customer_id", StringType(), True),
    StructField("product_id", StringType(), True),
    StructField("quantity", IntegerType(), True),
    StructField("unit_price", DecimalType(10, 2), True),
    StructField("order_status", StringType(), True),
    StructField("order_date", DateType(), True),
])
orders = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")
paid = orders.filter(col("order_status") == "paid").withColumn("order_amount", col("quantity") * col("unit_price"))
customers = spark.read.option("header", True).schema("customer_id string, region string").csv("/home/labex/project/data/customers.csv")
products = spark.read.option("header", True).schema("product_id string, category string").csv("/home/labex/project/data/products.csv")

with_customers = (
    paid.alias("o")
    .join(customers.alias("c"), col("o.customer_id") == col("c.customer_id"), "inner")
    .select(
        col("o.order_id").alias("order_id"), col("o.customer_id").alias("customer_id"),
        col("o.product_id").alias("product_id"), col("o.quantity").alias("quantity"),
        col("o.unit_price").alias("unit_price"), col("o.order_status").alias("order_status"),
        col("o.order_date").alias("order_date"), col("o.order_amount").alias("order_amount"),
        col("c.region").alias("region"),
    )
)
enriched = (
    with_customers.alias("e")
    .join(products.alias("p"), col("e.product_id") == col("p.product_id"), "inner")
    .select(
        col("e.order_id").alias("order_id"), col("e.customer_id").alias("customer_id"),
        col("e.product_id").alias("product_id"), col("e.quantity").alias("quantity"),
        col("e.unit_price").alias("unit_price"), col("e.order_status").alias("order_status"),
        col("e.order_date").alias("order_date"), col("e.order_amount").alias("order_amount"),
        col("e.region").alias("region"), col("p.category").alias("category"),
    )
)
category_sales = enriched.groupBy("category").agg(spark_sum("order_amount").alias("revenue"))
enriched.write.mode("overwrite").parquet("/home/labex/project/output/enriched-orders")
category_sales.write.mode("overwrite").parquet("/home/labex/project/output/category-sales")
enriched.orderBy("order_id").show(truncate=False)
category_sales.orderBy("category").show(truncate=False)
regional = enriched.groupBy("region").agg(spark_sum("order_amount").alias("revenue"))
office = enriched.filter(col("category") == "Office").groupBy("region").agg(spark_sum("order_amount").alias("revenue"))
regional.write.mode("overwrite").parquet("/home/labex/project/output/regional-sales")
office.write.mode("overwrite").parquet("/home/labex/project/output/office-regional-sales")
print("All categories by region:")
regional.orderBy("region").show(truncate=False)
print("Office category by region:")
office.orderBy("region").show(truncate=False)
spark.stop()
PYTHON

bigdata-python /home/labex/project/enrich_orders.py > /home/labex/project/output/regional-comparison.txt

cat /home/labex/project/output/regional-comparison.txt

预期得到预测中的全部分类和仅 Office 的地区结果。Office 各地区相加为 95.00，与前一步 Office 分类小计一致。全部地区相加为 125.00。

这些对账同时解释关联关系和所选业务子集。主 enriched Parquet 数据集仍有四笔类型正确的已付款订单，缺失匹配演示仍独立保存。接下来，你将通过 DataFrame 视图上的 Spark SQL 表达类似操作。

总结

你通过明确且唯一的客户键和商品键补充订单信息，选择无歧义输出列，并核对分类与地区总计。INNER JOIN 与 LEFT JOIN 对缺失匹配产生了有意不同的行为。接下来，你将对相同结构化数据比较 Spark SQL 与 DataFrame 表达式。

相关 Big Data 课程
Hadoop 入门

Hadoop 入门

bigdata
Hive 入门

Hive 入门

bigdatadatabasedata-science
Apache Kafka 入门

Apache Kafka 入门

pythonbigdata