Introdução
Você escreveu transformações do Spark, acionou ações e salvou dados. Agora relacionará essas operações conhecidas a evidências de execução: um plano que descreve o trabalho e um log real de eventos que registra jobs, stages e tasks. O objetivo é explicar o comportamento e ler erros úteis, em vez de memorizar contagens internas de tarefas.
O driver executa sua aplicação e coordena o trabalho. Em um cluster, executores executam tarefas nos recursos alocados; local[2] executa o trabalho do Spark nesta única máquina com duas threads de execução. Um trabalho (job) pode ser dividido em estágios (stages), e cada estágio executa tarefas (tasks) sobre partições de computação. Uma ação pode causar mais de um job, especialmente com planejamento de consultas e comportamento adaptativo; as contagens exatas podem variar.
Este laboratório independente fornece pedidos limpos, clientes, duas expressões deliberadamente incorretas e uma ferramenta de resumo de eventos somente de leitura. Você deve conhecer filtragem, agrupamento, junções e saídas Parquet. Você não ajustará recursos, não coletará o conjunto geral de dados em laços Python nem implantará um cluster. Use o Terminal.
Relacione uma consulta conhecida a seu plano de execução
Nesta etapa, você inspecionará uma consulta diária regional de vendas pagas antes de solicitar seus resultados. O leitor, o filtro, o valor, a junção de clientes e a agregação repetem habilidades anteriores.
Aqui, groupBy("order_date", "region") usa duas chaves juntas: cada combinação distinta de data e região forma um grupo. East em 14 de outubro é separado de East em 15 de outubro e de West em 14 de outubro. Contagens e somas são calculadas dentro de cada combinação. Isso reúne as perguntas anteriores de data e região em um relatório. explain(mode="formatted") imprime os operadores planejados em vez de retornar linhas do relatório.
A SparkSession também habilita um log real de eventos em output/spark-events. A compactação é desativada para tornar esse log didático legível após a aplicação parar. Essas configurações registram a execução; elas não substituem o registro por dados de jobs fabricados. O diretório de destino está preparado para você.
Comece e execute o script completo de inspeção:
cd /home/labex/project
cat > /home/labex/project/inspect_execution.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, count, sum as spark_sum, spark_partition_id
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType
spark = (
SparkSession.builder.appName("InspectRetailExecution").master("local[2]")
.config("spark.eventLog.enabled", "true")
.config("spark.eventLog.dir", "file:///home/labex/project/output/spark-events")
.config("spark.eventLog.compress", "false")
.config("spark.eventLog.rolling.enabled", "false")
.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")
customers = spark.read.option("header", True).schema("customer_id string, region string").csv("/home/labex/project/data/customers.csv")
paid = orders.filter(col("order_status") == "paid").withColumn("order_amount", col("quantity") * col("unit_price"))
matched = paid.alias("o").join(customers.alias("c"), col("o.customer_id") == col("c.customer_id"), "inner").select(
col("o.order_date").alias("order_date"), col("c.region").alias("region"), col("o.order_amount").alias("order_amount")
)
report = matched.groupBy("order_date", "region").agg(
count("*").alias("paid_order_count"), spark_sum("order_amount").alias("revenue")
)
print("Report execution plan:")
report.explain(mode="formatted")
print("Report action result:")
report.orderBy("order_date", "region").show(truncate=False)
report.write.mode("overwrite").parquet("/home/labex/project/output/daily-regional-sales")
print("Computation partition IDs represented by source rows:")
orders.select(spark_partition_id().alias("compute_partition")).dropDuplicates().orderBy("compute_partition").show(truncate=False)
spark.stop()
PYTHON
bigdata-python /home/labex/project/inspect_execution.py > /home/labex/project/output/execution-plan-and-result.txt
cat /home/labex/project/output/execution-plan-and-result.txt
Em Report execution plan, procure as funções de filtragem, projeção, junção, agregação e qualquer Exchange que mova registros entre partições de computação. Os nomes e a estrutura do plano podem variar com a otimização. Um Exchange é um limite de execução, não um novo diretório de saída.
Sob Report action result, espere 14 de outubro / East / 1 / 20.00, 14 de outubro / West / 1 / 25.00, 15 de outubro / East / 1 / 50.00 e 15 de outubro / West / 1 / 30.00. O total ainda é 125.00. show e o gravador solicitam a execução real do trabalho descrito; o planejamento e o tratamento de metadados de entrada também podem realizar trabalho preparatório.
A exibição final usa spark_partition_id para mostrar os IDs de partições de computação representados por esses registros de origem. dropDuplicates lista cada ID observado uma vez. Os IDs e a quantidade observada podem variar, e partições vazias podem não aparecer. Esses grupos de execução são diferentes dos diretórios nomeados por data gravados por partitionBy no laboratório anterior. Um organiza o trabalho; o outro define a estrutura dos dados armazenados.
Leia eventos reais de jobs, stages e tasks
Nesta etapa, você inspecionará o log de eventos produzido pela aplicação que acabou de executar. A ferramenta fornecida lê registros JSON de eventos desse diretório e resume eventos reais. Você não precisa escrever um analisador de logs nem selecionar um nome de arquivo específico de versão.
Execute a ferramenta de resumo, fornecendo o diretório de eventos como argumento de entrada. O > do shell guarda sua saída para revisão:
bigdata-python /home/labex/project/tools/summarize-events.py /home/labex/project/output/spark-events > /home/labex/project/output/event-summary.txt
cat /home/labex/project/output/event-summary.txt
Espere uma identidade real de aplicação, contagens positivas de Jobs started, Stages submitted e Tasks ended, além de referências de jobs a stages. Os números dependem desta execução e podem mudar com otimização, divisões de arquivos e versões do Spark. Eles são evidências observadas, não metas a reproduzir exatamente.
Uma ação solicita uma resposta; o Spark pode organizar o trabalho necessário em vários jobs. Um job faz referência a stages; stages contêm tasks para partições de trabalho. Limites de shuffle podem separar stages, enquanto várias transformações podem ser combinadas dentro de um stage. local[2] limita as threads simultâneas de execução; isso não significa que toda aplicação tenha exatamente dois jobs, stages ou tasks.
Para dados minúsculos, inicialização da JVM, agendamento e tratamento de arquivos podem ser perceptíveis em relação ao próprio cálculo. Uma pequena diferença de tempo decorrido não é uma conclusão confiável de ajuste. Use primeiro o plano e as evidências de eventos para entender o trabalho.
Diagnostique e corrija uma coluna desconhecida
Nesta etapa, você lerá um erro deliberado de resolução de nome antes de corrigir sua causa. Inspecione o exemplo incorreto preparado:
cat /home/labex/project/tools/broken-column.py
O esquema declara quantity, enquanto seu select final solicita quanity. Salve a saída real da falha. > redireciona a saída comum para o arquivo de log; 2>&1 envia a saída de erro padrão ao mesmo destino, salvando também o traceback. Espera-se que o próximo comando falhe.
bigdata-python /home/labex/project/tools/broken-column.py > /home/labex/project/output/column-error.log 2>&1
tail -n 20 /home/labex/project/output/column-error.log
Localize o nome de coluna não resolvido e quaisquer sugestões de nomes válidos. Um traceback longo contém detalhes de implementação, mas a referência à coluna aponta para a correção. A entrada não precisa de reinstalação nem de conversão para um laço Python.
Crie um programa corrigido com o nome de coluna declarado, deixando o exemplo incorreto de treinamento disponível para comparação:
cat > /home/labex/project/fixed_column.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("OrderExpressionExample").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")
fixed = orders.select("order_id", "quantity")
fixed.orderBy("order_id").show(truncate=False)
fixed.write.mode("overwrite").parquet("/home/labex/project/output/fixed-quantities")
spark.stop()
PYTHON
bigdata-python /home/labex/project/fixed_column.py > /home/labex/project/output/fixed-column-result.txt
cat /home/labex/project/output/fixed-column-result.txt
Espere os seis IDs com quantidades 2, 1, 1, 2, 1 e 2. A projeção agora é resolvida no esquema real e salva seu resultado efetivo. Nomes de colunas, incluindo sua grafia, fazem parte do contrato de dados.
Corrija um filtro que não é uma condição booleana
Nesta etapa, você distinguirá uma coluna inteira válida de uma condição válida de filtragem. Inspecione o segundo exemplo preparado:
cat /home/labex/project/tools/broken-filter.py
filter espera uma expressão booleana para cada linha. col("quantity") sozinho é um inteiro, não um teste de verdadeiro ou falso. Colete a incompatibilidade deliberada de tipo. Espera-se que o próximo comando falhe.
bigdata-python /home/labex/project/tools/broken-filter.py > /home/labex/project/output/filter-error.log 2>&1
tail -n 20 /home/labex/project/output/filter-error.log
Procure a condição de filtro e seu tipo inteiro não booleano. O reparo é expressar a comparação pretendida, quantity > 0, usando o operador de coluna já aprendido. Execute o programa completo corrigido:
cat > /home/labex/project/fixed_filter.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("OrderExpressionExample").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")
fixed = orders.filter(col("quantity") > 0)
fixed.orderBy("order_id").show(truncate=False)
fixed.write.mode("overwrite").parquet("/home/labex/project/output/fixed-orders")
spark.stop()
PYTHON
bigdata-python /home/labex/project/fixed_filter.py > /home/labex/project/output/fixed-filter-result.txt
cat /home/labex/project/output/fixed-filter-result.txt
As seis linhas limpas de exemplo têm quantidades positivas, portanto todas permanecem. Uma quantidade negativa ou zero falharia nesta condição; este reparo específico não aplica a regra separada de vendas pagas.
Agora você tem evidências em três níveis: a expressão e o esquema pretendidos, o trabalho real planejado e executado e resultados persistidos corretos após reparos focados. Mantenha a distinção entre diretórios de data armazenados e partições de computação ao explicar a execução. A seguir, submeterá um script independente de lote com parâmetros explícitos de entrada e saída.
Resumo
Você relacionou transformações conhecidas a um plano real e log de eventos, observou evidências de aplicação, jobs, stages e tasks sem fixar suas contagens e distinguiu partições de computação de diretórios de data. Corrigiu uma coluna não resolvida e um filtro não booleano usando o esquema e evidências de erros. A seguir, parâmetros explícitos de submissão tornarão sua aplicação em lote independente de um processo anterior.
