Introdução
Uma tabela de pedidos corretamente tipada é um ponto de partida, ainda não uma resposta. Uma pessoa da equipe pode querer apenas algumas colunas, somente pedidos pagos e um valor calculado a partir da quantidade e do preço unitário. Você expressará essas decisões como operações de DataFrame do Spark, mantendo os registros originais disponíveis.
Uma expressão de coluna descreve um cálculo ou uma condição para o Spark aplicar às linhas. Você selecionará campos, combinará condições, adicionará um valor decimal e dará um nome legível a uma coluna de resultado. Essas operações criam novos DataFrames; ações como show solicitam resultados reais. A distinção ajuda a explicar a execução preguiçosa do Spark sem precisar ainda dos detalhes internos de agendamento.
Este laboratório fornece de forma independente seis pedidos limpos de varejo. Você deve compreender o esquema CSV explícito do laboratório anterior e Python básico. O leitor completo e a SparkSession estão incluídos em cada versão executável, portanto nenhum processo nem arquivo anterior é necessário. Use o Terminal e o ambiente preparado bigdata-python.
Selecione as colunas necessárias para uma pergunta
Nesta etapa, você reduzirá as colunas da tabela mantendo os seis registros de pedidos. Comece pela origem:
cd /home/labex/project
cat /home/labex/project/data/orders.csv
select escolhe quais colunas um novo DataFrame expõe e em qual ordem. Ele não filtra linhas. O script completo abaixo repete o leitor explícito conhecido e seleciona o identificador do pedido, a quantidade, o preço e o status. Os campos de data e cliente permanecem em df, embora selected não os exponha.
Salve o script com um here-document entre aspas e execute-o. > grava o arquivo ou substitui a versão anterior; o marcador final PYTHON encerra o texto-fonte:
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 os seis IDs de pedidos com quatro colunas exibidas. orderBy torna a exibição previsível, enquanto show aciona o trabalho necessário para retornar linhas. Esta é uma seleção de colunas, às vezes chamada de projeção. Nenhum arquivo de origem foi alterado e nenhum pedido foi removido.
Combine condições para selecionar linhas
Nesta etapa, você manterá pedidos pagos com quantidade de pelo menos dois. col("order_status") se refere a uma coluna do Spark; compará-la com == "paid" cria uma condição para o Spark, em vez de um único True ou False do Python. >= cria a condição de quantidade.
Use & para combinar condições de colunas do Spark com a semântica de AND. Coloque parênteses ao redor de cada comparação porque a precedência de operadores do Python importa. O and comum do Python não é o operador para combinar objetos Column do Spark.
Preveja os IDs que atendem às condições: o1, o4 e o6 atendem a ambas. o2 está pago, mas tem quantidade um. Substitua e execute o 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
O resultado deverá conter apenas esses três IDs, cada um com quantidade 2. filter muda as linhas selecionadas; select controla as colunas exibidas. Você pode combinar ambas as operações enquanto df continua descrevendo a origem completa de seis linhas.
O limite de quantidade desta etapa é uma condição de demonstração, não uma definição geral de vendas. O próximo cálculo incluirá os quatro pedidos pagos, incluindo o2, pois cada pedido pago contribui para a receita.
Calcule e nomeie valores decimais de pedidos
Nesta etapa, você calculará o valor de cada pedido pago como quantidade multiplicada pelo preço unitário decimal. withColumn adiciona uma coluna nomeada a um novo DataFrame ou substitui essa coluna se seu nome já existir. Aqui, order_amount é calculado usando a expressão de multiplicação interna do Spark.
O select a seguir mantém o ID do pedido e dá à coluna calculada o nome de exibição amount usando alias. Um alias altera o nome da coluna do resultado; ele não edita o cabeçalho CSV. Os quatro pedidos pagos pertencem a este cálculo; linhas pendentes e reembolsadas são excluídas, sem tratar reembolsos como receita negativa nesta regra simplificada.
Execute esta versão 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 e o6/30.00. Zeros finais extras podem aparecer. O esquema deverá identificar amount como um tipo decimal. O Spark pode ampliar a precisão decimal durante a multiplicação; o ponto importante é que o cálculo mantém a aritmética decimal em vez de converter preços para ponto flutuante binário.
filter, withColumn, select e orderBy descrevem transformações e retornam novos DataFrames. Definir esse encadeamento não materializa imediatamente todas as tabelas intermediárias. show é uma ação que pede ao Spark para executar o trabalho necessário e exibir uma resposta. printSchema descreve a estrutura das colunas. Mais adiante, você inspecionará o plano real de execução por trás de um encadeamento como esse.
Mude um status e preveja seu efeito
Nesta etapa, você manterá o resultado de pagos e adicionará uma visualização separada de prática com pedidos pendentes. Preveja sua saída antes de executar: apenas o3 está pendente, com quantidade um e preço 10.00, portanto seu valor é 10.00.
O novo pending_result usa o mesmo encadeamento de expressões, alterando apenas o valor do status. Ele começa em df, não no paid_result já filtrado, portanto os registros pendentes continuam disponíveis:
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
Sob Paid orders, os mesmos quatro valores de pagos deverão permanecer. Sob Pending practice, espere apenas o3/10.00. Filtrar a visualização de pagos pelo status pendente não retornaria linhas porque esses registros já foram excluídos; começar pela origem adequada importa.
Ambos os encadeamentos são avaliados pelas expressões internas do Spark. Você não coletou o conjunto geral de dados em uma lista Python para percorrê-lo em um laço. Isso mantém a descrição de processamento adequada ao mecanismo de execução do Spark. Seu script final mantém o cálculo de pagos e a pequena comparação de prática, enquanto o CSV original ainda contém os seis pedidos.
Resumo
Você selecionou colunas, combinou condições de linhas com expressões do Spark, calculou valores decimais de pedidos e nomeou colunas de saída. Comparar visualizações de pagos e pendentes mostrou por que transformações devem partir da origem adequada e como ações solicitam resultados. A seguir, aplicará ideias conhecidas de leitura e seleção a uma entrada JSON.
