Введение
Заказ содержит идентификаторы клиента и товара, а отдельные справочники описывают, где живёт клиент и к какой категории относится товар. Соединение (join) связывает записи по этим ключам, чтобы отчёт отвечал на вопросы о выручке по регионам и категориям товаров.
Вы дополните оплаченные заказы по одному справочнику за раз, выберете однозначные выходные столбцы и сверите итоги. Также вы сравните INNER и 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 было бы неверной связью для этой таблицы. Сохраните промежуточные обогащённые поля заказов и выберите только категорию из справочника товаров.
Полная версия сохраняет первое соединение и добавляет второе. Явная проекция оставляет один product_id, а не два одинаково названных ключа. Затем знакомые groupBy и агрегатная сумма рассчитывают выручку категорий:
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 — на c_missing, отсутствующего в customers. Сначала предположите 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 остаётся доступен, хотя правый region отсутствует. NULL — отсутствие совпавшего региона, а не регион с буквальным именем NULL.
INNER JOIN подходит отчёту, полнота совпадений которого уже установлена. LEFT JOIN может сохранять неразрешённые записи для контроля качества или осознанной бизнес-политики. Выбор типа соединения меняет сохраняемые записи, а не просто оформление вывода.
Используйте обогащённые поля для регионального сравнения
На этом шаге вы примените изученные фильтрацию и группировку к основному обогащённому набору. Региональная выручка всех категорий должна быть East 70.00 и West 55.00. Предположите результат только Office: East останется 70.00, а West станет 25.00, поскольку заказ Home на 30.00 исключён.
Запустите полный основной скрипт с двумя дополнительными производными отчётами. Соединения остаются прежними; условие категории меняет только то, какие обогащённые строки заказов поступают во вторую агрегацию:
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.
Эти сверки объясняют и связь соединения, и выбранное бизнес-подмножество. Основной обогащённый набор Parquet по-прежнему содержит четыре правильно типизированных оплаченных заказа, а демонстрация отсутствия совпадения остаётся отдельной. Далее вы выразите сопоставимые операции через Spark SQL над представлением DataFrame.
Итоги
Вы дополнили заказы по явным уникальным ключам клиентов и товаров, выбрали однозначные выходные столбцы и сверили итоги категорий и регионов. INNER и LEFT JOIN намеренно дали разное поведение при отсутствии совпадения. Далее вы сравните Spark SQL и выражения DataFrame над теми же структурированными данными.
