Introducción
Un pedido conoce sus identificadores de cliente y producto, mientras que tablas de referencia separadas describen dónde vive el cliente y a qué categoría pertenece el producto. Una unión (join) conecta registros mediante esas claves para que un informe pueda responder preguntas como los ingresos por región o categoría de producto.
Enriquecerá los pedidos pagados con una tabla de consulta a la vez, seleccionará columnas de salida inequívocas y conciliará totales. También comparará INNER y LEFT JOIN cuando falte un cliente. Las claves de clientes y productos proporcionadas son únicas; cada pedido coincidente debería encontrar un registro en cada tabla de consulta. Las claves duplicadas podrían multiplicar filas unidas e inflar los ingresos, por lo que importa la relación prevista.
Este laboratorio independiente prepara pedidos limpios, dos clientes, tres productos y un ejemplo separado sin coincidencia. Debe conocer lecturas con tipos, filtros, importes decimales, agregación y salida Parquet mínima. No se requiere Hive, SQL, Hadoop, pandas ni configuración de clúster. Utilice Terminal.
Añada regiones de clientes mediante una clave explícita
En este paso, unirá los cuatro pedidos pagados con sus registros únicos de clientes. Examine la relación de origen:
cd /home/labex/project
cat /home/labex/project/data/orders.csv
cat /home/labex/project/data/customers.csv
Varios pedidos pueden compartir un customer_id, mientras que la tabla de consulta contiene una fila para c1 y una para c2. Los identificadores iguales conectan registros; emparejar posiciones de fila no sería una regla válida. c1 proporciona East y c2 proporciona West.
join de DataFrame acepta el otro DataFrame, una condición de coincidencia y un tipo de unión. inner conserva únicamente registros coincidentes. Los alias de DataFrame alias("o") y alias("c") dan a las entradas nombres cortos de referencia. col("o.customer_id") identifica la columna del pedido; col("c.customer_id") identifica la columna de la tabla de consulta.
Ambas entradas contienen una columna customer_id. El select explícito elige la clave del pedido y cada campo deseado, evitando nombres duplicados ambiguos. Un alias de DataFrame nombra una referencia de entrada; un alias de columna nombra un campo de resultado. Guarde y ejecute esta versión completa, cuyo lector y cálculo de importe repiten 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 cuatro filas pagadas enriquecidas. o1 y o4 deberían ser East, o2 y o6 West; sus importes permanecen en 20.00, 25.00, 50.00 y 30.00. Cuatro entradas pagadas con coincidencias únicas completas producen cuatro filas unidas. La salida Parquet conserva los campos con tipos y añade region sin cambiar el CSV original.
Añada categorías de productos y concilie ingresos
En este paso, añadirá una segunda tabla de consulta mediante product_id. Examine la correspondencia de productos:
cat /home/labex/project/data/products.csv
p1 y p2 pertenecen a Office (oficina); p3 pertenece a Home (hogar). Unir mediante customer_id sería la relación equivocada para esta tabla. Conserve los campos intermedios de pedidos enriquecidos y seleccione únicamente la categoría de la tabla de productos.
Esta versión completa conserva la primera unión y añade la segunda. La proyección explícita conserva un product_id en lugar de dejar dos claves con el mismo nombre. Después, los conocidos groupBy y la suma agregada calculan ingresos por categoría:
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
El resultado enriquecido debería seguir conteniendo cuatro pedidos pagados, ahora con región y categoría. El informe de categorías debería mostrar Home 30.00 y Office 95.00. Su total es el mismo ingreso pagado de 125.00.
Que el número de filas permanezca en cuatro y que los ingresos concilien proporciona dos comprobaciones útiles. Ninguno debería aumentar solo por añadir columnas descriptivas. Las claves de este ejemplo son únicas y completas; el siguiente paso explora una coincidencia ausente por separado en lugar de cambiar los totales principales de ventas.
Compare INNER y LEFT JOIN con un cliente ausente
En este paso, observará qué hace cada tipo de unión cuando la tabla de consulta no puede encontrar coincidencia. Examine el ejemplo independiente:
cat /home/labex/project/data/example-orders.csv
demo1 hace referencia a c1, mientras que demo2 hace referencia a c_missing, que está ausente de customers. Prediga primero INNER JOIN: solo debería permanecer demo1. LEFT JOIN conserva todos los registros de ejemplo del lado izquierdo y proporciona NULL para los campos que habrían procedido de un registro ausente del lado derecho.
Guarde un pequeño programa separado para que el script principal de ventas enriquecidas y sus datos permanezcan sin cambios:
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 debería contener demo1/c1/East. Left matches debería contener esa fila más demo2/c_missing/NULL. El customer_id del lado izquierdo sigue disponible aunque falte la región del lado derecho. NULL es la ausencia de una región coincidente, no una región llamada literalmente NULL.
INNER JOIN puede ser adecuado para un informe cuyas coincidencias completas ya se han establecido. LEFT JOIN puede conservar registros sin resolver para comprobaciones de calidad o una política comercial deliberada. Elegir el tipo de unión cambia qué registros sobreviven; no es simplemente una preferencia de presentación.
Utilice los campos enriquecidos para una comparación regional
En este paso, aplicará el filtrado y la agrupación ya aprendidos al conjunto principal enriquecido. Los ingresos regionales de todas las categorías deberían ser East 70.00 y West 55.00. Prediga el resultado solo de Office: East permanece en 70.00, mientras que West pasa a 25.00 porque se excluye su pedido Home de 30.00.
Ejecute el script principal completo con dos informes derivados adicionales. Sus uniones siguen siendo las mismas; la condición de categoría cambia únicamente qué filas enriquecidas de pedidos entran en la segunda agregación:
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 los resultados regionales previstos de todas las categorías y solo de Office. Las regiones de Office vuelven a sumar 95.00, coincidiendo con el subtotal de categoría Office del paso anterior. Las regiones globales vuelven a sumar 125.00.
Estas conciliaciones explican tanto la relación de unión como el subconjunto comercial seleccionado. Su conjunto principal Parquet enriquecido sigue conteniendo los cuatro pedidos pagados con tipos correctos y la demostración sin coincidencia permanece separada. A continuación, expresará operaciones comparables mediante Spark SQL sobre una vista de DataFrame.
Resumen
Enriqueció pedidos mediante claves explícitas y únicas de clientes y productos, seleccionó columnas de salida inequívocas y concilió totales por categoría y región. INNER y LEFT JOIN produjeron deliberadamente comportamientos diferentes ante coincidencias ausentes. A continuación, comparará Spark SQL y las expresiones de DataFrame sobre los mismos datos estructurados.
