PySpark 주문에 고객 및 상품 조회 데이터 조인하기

소개

주문에는 고객과 상품 식별자가 있고, 별도의 참조 테이블은 고객이 사는 지역과 상품이 속한 분류를 설명합니다. **조인(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 출력은 자료형을 유지하고 원본 CSV를 바꾸지 않은 채 region을 추가합니다.

상품 분류를 추가하고 매출 맞춰 보기

이 단계에서는 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 JOIN과 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이 있어야 합니다. 오른쪽 region이 없어도 왼쪽 customer_id는 남습니다. NULL은 이름이 NULL인 지역이 아니라 일치한 지역이 없다는 뜻입니다.

완전한 일치를 이미 확인한 보고서에는 INNER JOIN이 적절할 수 있습니다. LEFT JOIN은 품질 검사나 의도한 업무 정책에 따라 미해결 기록을 유지할 수 있습니다. 조인 유형 선택은 단순한 표시 취향이 아니라 살아남는 기록을 바꿉니다.

추가한 필드로 지역별 비교하기

이 단계에서는 주 데이터셋에 이미 배운 필터링과 그룹화를 적용합니다. 모든 분류의 지역 매출은 East 70.00, West 55.00이어야 합니다. Office만의 결과를 예상하세요. East는 70.00을 유지하고 West는 Home 주문 30.00이 제외되어 25.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 데이터셋에는 여전히 올바른 자료형의 결제 완료 주문 네 건이 있고 미일치 시연은 별개로 남습니다. 다음에는 DataFrame 뷰 위의 Spark SQL로 비슷한 연산을 표현합니다.

요약

명시적인 고유 고객 및 상품 키로 주문에 정보를 추가하고 모호하지 않은 출력 열을 선택했으며 분류 및 지역 합계를 맞춰 보았습니다. INNER JOIN과 LEFT JOIN은 의도적으로 서로 다른 미일치 동작을 보였습니다. 다음에는 같은 구조화 데이터에서 Spark SQL과 DataFrame 표현식을 비교합니다.

Spark 주문에 상품 및 고객 데이터 조인하기

소개

명시적 키로 고객 지역 추가하기

상품 분류를 추가하고 매출 맞춰 보기

고객 미일치에서 INNER JOIN과 LEFT JOIN 비교하기

추가한 필드로 지역별 비교하기

요약