Introdução
A empresa varejista já tem um conjunto correto e tipado de resultados de pedidos pagos. Agora, outra aplicação precisa ler esses resultados mais tarde, selecionar uma data e confiar que uma exportação repetida não tenha duplicado pedidos silenciosamente. Salvar dados faz parte do contrato de processamento, não é apenas a última linha de um script.
Você gravará Parquet com particionamento de diretórios por order_date. A raiz do conjunto conterá subdiretórios nomeados por data, e o Spark reconstruirá a coluna de data ao ler essa raiz. Esses diretórios de armazenamento são diferentes das partições de computação usadas para executar trabalho; o próximo laboratório examina a execução separadamente.
Este laboratório prepara de forma independente quatro pedidos pagos com valores decimais em uma entrada Parquet local. Você deve conhecer leituras de DataFrame, filtros e o fluxo mínimo de Parquet com sobrescrita. Não é necessário catálogo Hive, HDFS, conhecimento prévio de SQL, instalação nem saída anterior. Use o Terminal.
Grave um conjunto particionado por data e leia-o de volta
Nesta etapa, você organizará os resultados preparados em diretórios de data. O conjunto de entrada em data/prepared-orders contém quatro pedidos, com os mesmos tipos de quantidade, preço decimal, data e order_amount decimal usados anteriormente. Parquet é binário; inspecione-o pelo Spark em vez de esperar que cat exiba texto CSV.
Comece no espaço de trabalho:
cd /home/labex/project
partitionBy("order_date") instrui o gravador a colocar linhas em diretórios nomeados por data. mode("overwrite") substitui o conjunto derivado existente no caminho de saída especificado. Mantenha esse caminho de saída separado de input_path, que deve permanecer intacto.
Salve e execute o programa completo:
cat > /home/labex/project/write_sales_data.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("WriteSalesData").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
input_path = "/home/labex/project/data/prepared-orders"
output_path = "/home/labex/project/output/orders-by-date"
prepared = spark.read.parquet(input_path)
prepared.write.mode("overwrite").partitionBy("order_date").parquet(output_path)
print("Main read-back:")
reloaded = spark.read.parquet(output_path)
reloaded.orderBy("order_id").show(truncate=False)
reloaded.printSchema()
print("Main rows:", reloaded.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/write_sales_data.py > /home/labex/project/output/partitioned-read-back.txt
cat /home/labex/project/output/partitioned-read-back.txt
Espere quatro registros reconstruídos: o1 e o2 em 14 de outubro, e o4 e o6 em 15 de outubro, com valores 20.00, 25.00, 50.00 e 30.00. Main rows deverá ser 4. Confira se quantity é inteiro, preço e valor são decimais e order_date é data.
Uma coluna de partição pode aparecer em outra posição após a leitura porque o Spark a reconstrói a partir dos nomes de diretório em vez de armazená-la como um campo comum em cada arquivo de partição. Nomes e valores de colunas estabelecem sua identidade; use um select explícito quando um consumidor posterior exigir uma ordem específica de colunas. A ordem de exibição também exige orderBy.
Inspecione diretórios e selecione uma data
Nesta etapa, você relacionará a estrutura de diretórios à coluna de data reconstruída. Liste a saída recursivamente; -R inclui os subdiretórios de data e seus arquivos:
ls -R /home/labex/project/output/orders-by-date
Espere os diretórios order_date=2026-10-14 e order_date=2026-10-15. Os nomes de arquivos part, sua quantidade e os metadados ocultos podem variar; não fixe um arquivo part como se fosse o conjunto inteiro. A raiz significativa é orders-by-date.
Ler essa raiz permite ao Spark descobrir os valores de partição. Essa descoberta da estrutura de arquivos não exige registrar uma partição nem tabela no Hive. O script independente a seguir lê a raiz, compara sua coluna DateType com uma string válida de data ISO e salva separadamente os registros do dia selecionado:
cat > /home/labex/project/inspect_sales_data.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
spark = SparkSession.builder.appName("InspectDateDataset").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
root = "/home/labex/project/output/orders-by-date"
df = spark.read.parquet(root)
selected = df.filter(col("order_date") == "2026-10-14")
selected.orderBy("order_id").show(truncate=False)
selected.write.mode("overwrite").parquet("/home/labex/project/output/selected-day")
spark.stop()
PYTHON
bigdata-python /home/labex/project/inspect_sales_data.py > /home/labex/project/output/selected-date.txt
cat /home/labex/project/output/selected-date.txt
Espere apenas o1 e o2, com suas datas de 14 de outubro e valores 20.00 e 25.00. A outra data ainda existe no conjunto principal. Uma leitura filtrada muda os registros retornados; ela não exclui diretórios não selecionados.
Você inspecionou a organização do armazenamento e o resultado real da consulta. Nesta amostra minúscula, o tempo decorrido não estabelece que um formato ou uma estrutura seja mais rápida. O objetivo é um conjunto correto e detectável que leitores posteriores possam usar.
Compare sobrescritas repetidas com acréscimo de forma segura
Nesta etapa, você preservará o conjunto principal correto e experimentará em um diretório separado. Preveja as contagens descartáveis: gravar as mesmas quatro linhas com sobrescrita duas vezes deixa quatro linhas em cada vez; acrescentar uma cópia completa depois produz oito linhas. Essas linhas extras são exportações duplicadas, não novos pedidos.
Execute esta versão completa. O caminho principal usa sobrescrita. Apenas write-mode-demo recebe o acréscimo deliberado:
cat > /home/labex/project/write_sales_data.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("WriteSalesData").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
input_path = "/home/labex/project/data/prepared-orders"
output_path = "/home/labex/project/output/orders-by-date"
prepared = spark.read.parquet(input_path)
prepared.write.mode("overwrite").partitionBy("order_date").parquet(output_path)
print("Main read-back:")
reloaded = spark.read.parquet(output_path)
reloaded.orderBy("order_id").show(truncate=False)
reloaded.printSchema()
print("Main rows:", reloaded.count())
practice_path = "/home/labex/project/output/write-mode-demo"
prepared.write.mode("overwrite").partitionBy("order_date").parquet(practice_path)
print("First overwrite rows:", spark.read.parquet(practice_path).count())
prepared.write.mode("overwrite").partitionBy("order_date").parquet(practice_path)
print("Second overwrite rows:", spark.read.parquet(practice_path).count())
prepared.write.mode("append").partitionBy("order_date").parquet(practice_path)
print("After append rows:", spark.read.parquet(practice_path).count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/write_sales_data.py > /home/labex/project/output/write-mode-comparison.txt
cat /home/labex/project/output/write-mode-comparison.txt
Procure Main rows: 4, First overwrite rows: 4, Second overwrite rows: 4 e After append rows: 8. O conjunto descartável agora contém intencionalmente duas cópias de cada pedido, mostrando por que acrescentar um retrato completo é inadequado para a exportação principal.
Execute o programa salvo novamente para conferir a repetibilidade do fluxo inteiro:
bigdata-python /home/labex/project/write_sales_data.py > /home/labex/project/output/write-mode-comparison.txt
cat /home/labex/project/output/write-mode-comparison.txt
As mesmas contagens deverão aparecer. Cada execução completa reconstrói o conjunto principal e reinicia o experimento descartável antes de seu acréscimo intencional, portanto ele não cresce para doze nem dezesseis linhas. Acrescentar é uma política distinta para dados realmente novos e sem sobreposição; a sobrescrita é a política escolhida para este retrato derivado completo.
O conjunto original prepared-orders continua disponível. Diretórios de data descrevem valores armazenados, enquanto as partições de computação do Spark em execução descrevem como um trabalho é dividido. A seguir, você relacionará esse segundo significado de partição a planos e execução real.
Resumo
Você gravou um conjunto Parquet particionado por data, reconstruiu seus registros tipados a partir da raiz e selecionou uma data sem remover as outras. Sobrescritas repetidas preservaram o retrato principal, enquanto um acréscimo descartável demonstrou cópias duplicadas. A seguir, planos de execução e eventos reais explicarão partições de computação, jobs, stages e tasks.
