Introducción
Una tabla de pedidos con tipos correctos es un punto de partida, todavía no una respuesta. Un colega puede querer solo unas pocas columnas, únicamente los pedidos pagados y un importe calculado a partir de la cantidad y el precio unitario. Expresará esas decisiones como operaciones de DataFrame de Spark mientras mantiene disponibles los registros originales.
Una expresión de columna describe un cálculo o una condición que Spark debe aplicar a las filas. Seleccionará campos, combinará condiciones, añadirá un importe decimal y dará un nombre legible a una columna de resultado. Estas operaciones crean DataFrames nuevos; acciones como show solicitan resultados reales. La distinción ayuda a explicar la ejecución diferida de Spark sin necesitar todavía los detalles internos de planificación.
Este laboratorio proporciona de forma independiente seis pedidos limpios de ventas. Debe comprender el esquema CSV explícito del laboratorio anterior y Python básico. Cada versión ejecutable incluye el lector completo y la SparkSession, por lo que no se necesita ningún proceso ni archivo anterior. Utilice Terminal y el entorno preparado bigdata-python.
Seleccione las columnas necesarias para una pregunta
En este paso, reducirá las columnas de la tabla conservando los seis registros de pedidos. Comience con el origen:
cd /home/labex/project
cat /home/labex/project/data/orders.csv
select elige qué columnas expone un DataFrame nuevo y en qué orden. No filtra filas. El script completo siguiente repite el lector explícito conocido y después selecciona el identificador de pedido, la cantidad, el precio y el estado. Los campos de fecha y cliente permanecen en df, aunque selected no los exponga.
Guarde el script con un documento en línea entre comillas y ejecútelo. > escribe el archivo o sustituye la versión anterior; el marcador PYTHON final termina el texto del código:
cat > /home/labex/project/transform_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType
spark = SparkSession.builder.appName("TransformOrders").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),
])
df = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")
selected = df.select("order_id", "quantity", "unit_price", "order_status")
selected.orderBy("order_id").show(truncate=False)
spark.stop()
PYTHON
bigdata-python /home/labex/project/transform_orders.py > /home/labex/project/output/selected-orders.txt
cat /home/labex/project/output/selected-orders.txt
Espere los seis ID de pedido con cuatro columnas mostradas. orderBy hace predecible la visualización, mientras que show activa el trabajo necesario para devolver filas. Esto es una selección de columnas, a veces denominada proyección. No se cambió ningún archivo de origen ni se eliminó ningún pedido.
Combine condiciones para seleccionar filas
En este paso, conservará pedidos pagados con una cantidad de al menos dos. col("order_status") hace referencia a una columna de Spark; compararla con == "paid" construye una condición para Spark, en lugar de un único True o False de Python. >= construye la condición de cantidad.
Utilice & para combinar condiciones de columnas de Spark con semántica AND. Ponga paréntesis alrededor de cada comparación porque importa la precedencia de operadores de Python. El and ordinario de Python no es el operador para combinar objetos Column de Spark.
Prediga los ID que cumplen: o1, o4 y o6 cumplen ambas condiciones. o2 está pagado, pero tiene cantidad uno. Sustituya y ejecute el script completo:
cat > /home/labex/project/transform_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType
spark = SparkSession.builder.appName("TransformOrders").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),
])
df = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")
eligible = df.filter((col("order_status") == "paid") & (col("quantity") >= 2))
eligible.select("order_id", "quantity", "unit_price", "order_status").orderBy("order_id").show(truncate=False)
spark.stop()
PYTHON
bigdata-python /home/labex/project/transform_orders.py > /home/labex/project/output/eligible-orders.txt
cat /home/labex/project/output/eligible-orders.txt
El resultado debería contener únicamente esos tres ID, cada uno con cantidad 2. filter cambia las filas seleccionadas; select controla las columnas mostradas. Puede combinar ambas operaciones mientras df sigue describiendo el origen completo de seis filas.
El umbral de cantidad de este paso es una condición de demostración, no una definición general de ventas. El siguiente cálculo incluirá los cuatro pedidos pagados, incluido o2, porque cada pedido pagado contribuye a los ingresos.
Calcule y nombre importes decimales de pedidos
En este paso, calculará el importe de cada pedido pagado como cantidad multiplicada por precio unitario decimal. withColumn añade una columna con nombre a un DataFrame nuevo, o sustituye esa columna si su nombre ya existe. Aquí order_amount se calcula mediante la expresión incorporada de multiplicación de Spark.
El select siguiente conserva el ID de pedido y da a la columna calculada el nombre de presentación amount mediante alias. Un alias cambia el nombre de columna del resultado; no edita el encabezado CSV. Los cuatro pedidos pagados pertenecen a este cálculo; se excluyen las filas pendientes y reembolsadas, sin tratar los reembolsos como ingresos negativos en esta regla simplificada.
Ejecute esta versión completa:
cat > /home/labex/project/transform_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType
spark = SparkSession.builder.appName("TransformOrders").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),
])
df = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")
paid_result = (
df.filter(col("order_status") == "paid")
.withColumn("order_amount", col("quantity") * col("unit_price"))
.select("order_id", col("order_amount").alias("amount"))
)
print("Paid orders:")
paid_result.orderBy("order_id").show(truncate=False)
paid_result.printSchema()
spark.stop()
PYTHON
bigdata-python /home/labex/project/transform_orders.py > /home/labex/project/output/paid-amounts.txt
cat /home/labex/project/output/paid-amounts.txt
Espere o1/20.00, o2/25.00, o4/50.00 y o6/30.00. Pueden aparecer ceros finales adicionales. El esquema debería identificar amount como un tipo decimal. Spark puede ampliar la precisión decimal durante la multiplicación; lo importante es que el cálculo conserva la aritmética decimal en lugar de convertir los precios a punto flotante binario.
filter, withColumn, select y orderBy describen transformaciones y devuelven DataFrames nuevos. Definir esta cadena no materializa inmediatamente todas las tablas intermedias. show es una acción que pide a Spark ejecutar el trabajo necesario y mostrar una respuesta. printSchema describe la estructura de columnas. Más adelante examinará el plan real de ejecución detrás de una cadena así.
Cambie un estado y prediga su efecto
En este paso, conservará el resultado pagado y añadirá una vista separada de práctica de pedidos pendientes. Prediga su salida antes de ejecutar: solo o3 está pendiente, con cantidad uno y precio 10.00, por lo que su importe es 10.00.
El nuevo pending_result utiliza la misma cadena de expresiones cambiando solo el valor del estado. Comienza desde df, no desde el paid_result ya filtrado, por lo que los registros pendientes siguen disponibles:
cat > /home/labex/project/transform_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType
spark = SparkSession.builder.appName("TransformOrders").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),
])
df = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")
paid_result = (
df.filter(col("order_status") == "paid")
.withColumn("order_amount", col("quantity") * col("unit_price"))
.select("order_id", col("order_amount").alias("amount"))
)
print("Paid orders:")
paid_result.orderBy("order_id").show(truncate=False)
paid_result.printSchema()
pending_result = (
df.filter(col("order_status") == "pending")
.withColumn("order_amount", col("quantity") * col("unit_price"))
.select("order_id", col("order_amount").alias("amount"))
)
print("Pending practice:")
pending_result.orderBy("order_id").show(truncate=False)
spark.stop()
PYTHON
bigdata-python /home/labex/project/transform_orders.py > /home/labex/project/output/status-comparison.txt
cat /home/labex/project/output/status-comparison.txt
Bajo Paid orders (pedidos pagados), deberían permanecer los mismos cuatro importes pagados. Bajo Pending practice (práctica de pendientes), espere únicamente o3/10.00. Filtrar la vista pagada por estado pendiente no devolvería filas porque esos registros ya se excluyeron; importa comenzar desde el origen adecuado.
Ambas cadenas se evalúan mediante las expresiones incorporadas de Spark. No recopiló el conjunto de datos general en una lista de Python para recorrerlo con un bucle. Esto mantiene la descripción del procesamiento adecuada para el motor de ejecución de Spark. Su script final conserva el cálculo pagado y la pequeña comparación de práctica, mientras que el CSV original sigue conteniendo los seis pedidos.
Resumen
Seleccionó columnas, combinó condiciones de filas con expresiones de Spark, calculó importes decimales de pedidos y nombró columnas de salida. Comparar las vistas pagadas y pendientes mostró por qué las transformaciones deben comenzar desde el origen adecuado y cómo las acciones solicitan resultados. A continuación, aplicará las ideas conocidas de lectura y selección a la entrada JSON.
