Associe pedidos do PySpark a referências de clientes e produtos

Introdução

Um pedido contém seus identificadores de cliente e produto, enquanto tabelas de referência separadas descrevem onde o cliente mora e a qual categoria o produto pertence. Uma junção (join) conecta registros por essas chaves para que um relatório responda a perguntas como receita por região ou categoria de produto.

Você enriquecerá pedidos pagos com uma tabela de consulta por vez, selecionará colunas de saída sem ambiguidade e conciliará os totais. Também comparará INNER e LEFT JOIN quando um cliente estiver ausente. As chaves fornecidas de clientes e produtos são únicas; cada pedido correspondente deverá encontrar um registro em cada tabela de consulta. Chaves duplicadas nas referências poderiam multiplicar linhas associadas e inflar a receita, portanto a relação pretendida importa.

Este laboratório independente prepara pedidos limpos, dois clientes, três produtos e um exemplo separado de correspondência ausente. Você deve conhecer leituras tipadas, filtros, valores decimais, agregação e saída mínima em Parquet. Não é necessário Hive, SQL, Hadoop, pandas nem configurar um cluster. Use o Terminal.

Adicione regiões de clientes por uma chave explícita

Nesta etapa, você associará os quatro pedidos pagos a seus registros únicos de clientes. Inspecione a relação de origem:

cd /home/labex/project

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

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

Vários pedidos podem compartilhar um customer_id, enquanto a tabela de consulta contém uma linha para c1 e uma para c2. Identificadores iguais conectam registros; corresponder posições de linhas não seria uma regra válida. c1 fornece East e c2 fornece West.

O join do DataFrame aceita o outro DataFrame, uma condição de correspondência e um tipo de junção. inner mantém apenas registros correspondentes. Os aliases de DataFrame alias("o") e alias("c") dão nomes curtos de referência às entradas. col("o.customer_id") identifica a coluna do lado dos pedidos; col("c.customer_id") identifica a coluna do lado da referência.

Ambas as entradas contêm uma coluna customer_id. O select explícito escolhe a chave do lado dos pedidos e cada campo desejado, evitando nomes duplicados ambíguos. Um alias de DataFrame nomeia uma referência de entrada; um alias de coluna nomeia um campo de resultado. Salve e execute esta versão completa, cujo leitor e cálculo de valor repetem habilidades anteriores:

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

Espere quatro linhas enriquecidas de pedidos pagos. o1 e o4 deverão ter East, o2 e o6, West; seus valores continuam 20.00, 25.00, 50.00 e 30.00. Quatro entradas pagas com correspondências completas e únicas produzem quatro linhas associadas. A saída Parquet preserva os campos tipados e acrescenta region sem alterar o CSV bruto.

Adicione categorias de produtos e concilie a receita

Nesta etapa, você adicionará uma segunda referência usando product_id. Inspecione o mapeamento de produtos:

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

p1 e p2 pertencem a Office; p3 pertence a Home. Fazer a junção por customer_id seria a relação errada para esta tabela. Mantenha os campos intermediários do pedido enriquecido e selecione apenas category da referência de produtos.

Esta versão completa mantém a primeira junção e acrescenta a segunda. A projeção explícita mantém um product_id em vez de deixar duas chaves com nomes idênticos. Em seguida, o groupBy conhecido e a agregação sum calculam a receita por categoria:

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

O resultado enriquecido ainda deverá conter quatro pedidos pagos, agora com região e categoria. O relatório por categoria deverá mostrar Home 30.00 e Office 95.00. Seu total é a mesma receita paga de 125.00.

A contagem de linhas permanecendo em quatro e a receita conciliada fornecem duas conferências úteis. Nenhuma delas deve aumentar apenas porque colunas descritivas foram adicionadas. As chaves destes dados de exemplo são únicas e completas; a próxima etapa explora uma correspondência ausente separadamente em vez de alterar os totais principais de vendas.

Compare INNER e LEFT JOIN com um cliente ausente

Nesta etapa, você observará o que cada tipo de junção faz quando a referência não encontra correspondência. Inspecione o exemplo independente:

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

demo1 se refere a c1, enquanto demo2 se refere a c_missing, que está ausente de customers. Preveja primeiro o INNER JOIN: apenas demo1 deverá permanecer. LEFT JOIN mantém todos os registros de exemplo do lado esquerdo e fornece NULL para campos que viriam de um registro ausente do lado direito.

Salve um pequeno programa separado para que o script principal de vendas enriquecidas e seus dados permaneçam inalterados:

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 deverá conter demo1/c1/East. Left matches deverá conter essa linha mais demo2/c_missing/NULL. O customer_id do lado esquerdo continua disponível, embora a região do lado direito esteja ausente. NULL é a ausência de uma região correspondente, não uma região literalmente chamada NULL.

INNER JOIN pode ser adequado para um relatório cujas correspondências completas já foram estabelecidas. LEFT JOIN pode manter registros não resolvidos para conferências de qualidade ou uma política deliberada de negócio. Escolher o tipo de junção muda quais registros permanecem; não é apenas uma preferência de exibição.

Use os campos enriquecidos para uma comparação regional

Nesta etapa, você aplicará filtragem e agrupamento já aprendidos ao conjunto principal de dados enriquecidos. A receita regional de todas as categorias deverá ser East 70.00 e West 55.00. Preveja o resultado apenas de Office: East permanece 70.00, enquanto West passa a 25.00 porque seu pedido Home de 30.00 é excluído.

Execute o script principal completo com dois relatórios derivados adicionais. Suas junções permanecem iguais; a condição de categoria muda apenas quais linhas de pedidos enriquecidos entram na segunda agregação:

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

Espere os resultados regionais previstos de todas as categorias e apenas de Office. As regiões de Office voltam a somar 95.00, correspondendo ao subtotal da categoria Office da etapa anterior. As regiões gerais voltam a somar 125.00.

Essas conciliações explicam tanto a relação da junção quanto o subconjunto de negócio selecionado. Seu conjunto principal enriquecido em Parquet ainda contém os quatro pedidos pagos corretamente tipados, e a demonstração de correspondência ausente permanece separada. A seguir, você expressará operações comparáveis com Spark SQL sobre uma visualização de DataFrame.

Resumo

Você enriqueceu pedidos por chaves explícitas e únicas de clientes e produtos, selecionou colunas de saída sem ambiguidade e conciliou totais por categoria e região. INNER e LEFT JOIN produziram comportamentos deliberadamente diferentes para correspondências ausentes. A seguir, comparará Spark SQL e expressões de DataFrame sobre os mesmos dados estruturados.