Introdução
Um script pode ser executado com sucesso e ainda passar registros ruins a um relatório. O recebimento de varejo de hoje contém uma linha repetida de pedido, um cliente ausente, texto numérico inválido, uma quantidade não positiva, um preço negativo e uma data malformada. Preencher automaticamente esses problemas com zero inventaria informações de negócio.
Você separará três decisões: ler o texto sem perder as evidências originais, converter com segurança os campos numéricos e de data pretendidos e manter apenas as linhas que atendam a uma regra declarada. Em seguida, linhas válidas idênticas serão deduplicadas. Por fim, você salvará e lerá de volta um conjunto limpo de dados Parquet. Parquet é um formato binário colunar e tipado de arquivo; este laboratório ensina apenas o fluxo mínimo de gravação e leitura, deixando o armazenamento particionado para mais adiante.
O ambiente preparado fornece dois recebimentos independentes. Você deve conhecer leitura de CSV, expressões de coluna do Spark, filtragem e preços decimais. A expressão de conversão segura usada aqui é aceita pela versão preparada do Spark; você não precisa mudar sua configuração de tratamento de erros. Use o Terminal. Não é necessário usar arquivos de saída anteriores, Hadoop, Hive nem instalar serviços.
Preserve o recebimento como dados intermediários de string
Nesta etapa, você observará defeitos antes de converter valores. Inspecione a origem local:
cd /home/labex/project
cat /home/labex/project/data/dirty-orders.csv
Há quatorze linhas de negócio. Um DataFrame de staging, ou preparação intermediária, mantém esses campos de entrada como strings, permitindo ver valores como two, unknown e not-a-date sem fingir que são números ou datas válidos. O tratamento padrão do leitor CSV torna NULL um campo vazio de cliente; NULL representa um valor ausente, não o texto "NULL".
Salve e execute o programa completo de staging. Como antes, o here-document PYTHON entre aspas grava o fonte literalmente. show(30, truncate=False) permite exibir até trinta registros e mantém legível o texto dos campos:
cat > /home/labex/project/clean_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, trim, expr
spark = SparkSession.builder.appName("CleanOrderDelivery").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
staging = spark.read.option("header", True).csv("/home/labex/project/data/dirty-orders.csv")
staging.printSchema()
staging.orderBy("order_id").show(30, truncate=False)
print("Staging rows:", staging.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/clean_orders.py > /home/labex/project/output/staging-orders.txt
cat /home/labex/project/output/staging-orders.txt
Espere quatorze registros exibidos e Staging rows: 14. Todas as colunas deverão ser strings. Localize a quantidade two de b2 e o preço unknown de b4; são evidências úteis, não erros ocultados por um valor padrão adivinhado. o2 aparece duas vezes, e o status de o1 tem espaços ao redor.
O arquivo original é somente de leitura neste exercício. Você criará DataFrames derivados e diretórios de saída, deixando o recebimento bruto disponível para revisão.
Converta valores com segurança sem inventar valores padrão
Nesta etapa, você normalizará os dois campos de texto que precisam da remoção de espaços ao redor e tentará a conversão numérica e de data. trim remove espaços iniciais e finais. Ele não adivinha um cliente ausente nem altera o status de pagamento de um pedido.
expr permite fornecer uma expressão interna do Spark como texto. try_cast(quantity AS INT) tenta uma conversão para inteiro e produz NULL se o valor não puder ser convertido. As expressões equivalentes solicitam DECIMAL(10,2) para preços e DATE para datas. Conversões comuns com falha podem gerar erros sob o comportamento ANSI do Spark; esta conversão segura explícita evita depender de uma configuração global.
A mudança de esquema preserva as linhas para que problemas de conversão continuem observáveis. Execute esta versão:
cat > /home/labex/project/clean_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, trim, expr
spark = SparkSession.builder.appName("CleanOrderDelivery").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
staging = spark.read.option("header", True).csv("/home/labex/project/data/dirty-orders.csv")
parsed = (
staging
.withColumn("customer_id", trim(col("customer_id")))
.withColumn("order_status", trim(col("order_status")))
.withColumn("quantity", expr("try_cast(quantity AS INT)"))
.withColumn("unit_price", expr("try_cast(unit_price AS DECIMAL(10,2))"))
.withColumn("order_date", expr("try_cast(order_date AS DATE)"))
)
parsed.printSchema()
parsed.orderBy("order_id").show(30, truncate=False)
print("Parsed rows:", parsed.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/clean_orders.py > /home/labex/project/output/parsed-orders.txt
cat /home/labex/project/output/parsed-orders.txt
Espere Parsed rows: 14. A quantidade de b2, o preço de b4 e a data de b6 agora deverão ser NULL. A quantidade zero de b3 e o preço negativo de b5 são valores numéricos sintaticamente válidos, mas ainda violam as regras de negócio que você aplicará a seguir. O status de o1 agora deverá ser paid sem espaços ao redor.
Conversão bem-sucedida não é o mesmo que um pedido válido. Manter essas fases distintas permite explicar se uma linha falhou na conversão ou em uma restrição de negócio. Nenhum preço ausente foi substituído por zero; o pedido o7 com preço zero real continua distinguível de um preço desconhecido.
Aplique regras explícitas de validade e duplicatas
Nesta etapa, você manterá registros válidos de negócio e removerá apenas duplicatas idênticas. Use estas regras:
- customer_id está presente e não vazio após a remoção de espaços;
- quantity foi convertido com sucesso e é maior que zero;
- unit_price foi convertido com sucesso e é maior ou igual a zero;
- order_date foi convertido com sucesso;
- linhas idênticas nos sete campos normalizados são mantidas uma vez.
Os outros campos de identificadores dos dados de exemplo já estão bem formados. Não há versões conflitantes do mesmo ID de pedido, portanto este exercício não escolhe uma versão vencedora entre status em mudança. Registros pagos, pendentes e reembolsados permanecem se forem válidos nos demais aspectos; limpar um recebimento é diferente de escolher receita paga para um relatório.
isNotNull verifica explicitamente a presença. Uma comparação envolvendo NULL não se torna uma correspondência verdadeira, e filter mantém apenas condições verdadeiras. Combinar as condições com & deixa a política visível. dropDuplicates() sem subconjunto remove linhas completas idênticas; usar apenas order_id seria outra política, que poderia descartar informações conflitantes.
Execute esta versão:
cat > /home/labex/project/clean_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, trim, expr
spark = SparkSession.builder.appName("CleanOrderDelivery").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
staging = spark.read.option("header", True).csv("/home/labex/project/data/dirty-orders.csv")
parsed = (
staging
.withColumn("customer_id", trim(col("customer_id")))
.withColumn("order_status", trim(col("order_status")))
.withColumn("quantity", expr("try_cast(quantity AS INT)"))
.withColumn("unit_price", expr("try_cast(unit_price AS DECIMAL(10,2))"))
.withColumn("order_date", expr("try_cast(order_date AS DATE)"))
)
valid = parsed.filter(
col("customer_id").isNotNull()
& (col("customer_id") != "")
& col("quantity").isNotNull() & (col("quantity") > 0)
& col("unit_price").isNotNull() & (col("unit_price") >= 0)
& col("order_date").isNotNull()
)
clean = valid.dropDuplicates()
print("Clean rows:")
clean.orderBy("order_id").show(30, truncate=False)
print("Staging count:", staging.count())
print("Valid count before deduplication:", valid.count())
print("Clean count:", clean.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/clean_orders.py > /home/labex/project/output/cleaning-result.txt
cat /home/labex/project/output/cleaning-result.txt
Espere Staging count: 14, Valid count before deduplication: 8 e Clean count: 7. b1–b6 são excluídos pelas regras declaradas, enquanto uma repetição de o2 é removida. Os sete IDs mantidos são o1 a o7.
O preço zero real do pedido o7 é válido sob a regra de preço não negativo. o3 continua pendente e o5 reembolsado: nenhum desses status torna o registro de origem malformado. O conjunto resultante está limpo para esta política declarada, não é uma afirmação de que toda possível anomalia do mundo real foi resolvida.
Salve, leia de volta e teste um defeito extra
Nesta etapa, você tornará o conjunto limpo reutilizável e testará a mesma política em outro recebimento. write.mode("overwrite").parquet(path) grava um conjunto de dados Parquet em um diretório, substituindo o conteúdo derivado anterior nesse local. Mantenha esse diretório separado da entrada bruta. O Spark pode criar vários arquivos part e um marcador de conclusão; não dependa de um nome ou quantidade específica de arquivos.
spark.read.parquet(path) lê o conjunto salvo e seus tipos armazenados. Consultar esse DataFrame lido de volta confere o artefato real, em vez de simplesmente exibir o plano anterior à gravação. O programa completo abaixo coloca a lógica já aprendida em uma função Python para que ambos os recebimentos usem a mesma regra de limpeza.
def clean_orders(input_path): define a função e nomeia seu parâmetro de entrada. As linhas indentadas são seu corpo; elas são executadas quando a função é chamada. A primeira chamada fornece o nome do arquivo original, e a segunda fornece o nome do arquivo do recebimento extra. Dentro de cada chamada, input_path se refere ao valor fornecido. return clean devolve o DataFrame resultante a quem chamou; ele não salva arquivos. As linhas de gravação sem indentação realizam essa ação separada. Preserve a indentação ao copiar o programa completo.
O recebimento extra contém as quatorze linhas originais mais uma linha de quantidade negativa e outra cópia exata do pedido válido de preço zero. Preveja o resultado limpo: essas adições não deverão alterar os sete registros mantidos. Inspecione as adições:
tail -n 2 /home/labex/project/data/dirty-orders-extra.csv
Execute o programa completo de gravação e leitura de volta:
cat > /home/labex/project/clean_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, trim, expr
spark = SparkSession.builder.appName("CleanOrderDelivery").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
def clean_orders(input_path):
staging = spark.read.option("header", True).csv(input_path)
parsed = (
staging
.withColumn("customer_id", trim(col("customer_id")))
.withColumn("order_status", trim(col("order_status")))
.withColumn("quantity", expr("try_cast(quantity AS INT)"))
.withColumn("unit_price", expr("try_cast(unit_price AS DECIMAL(10,2))"))
.withColumn("order_date", expr("try_cast(order_date AS DATE)"))
)
valid = parsed.filter(
col("customer_id").isNotNull()
& (col("customer_id") != "")
& col("quantity").isNotNull() & (col("quantity") > 0)
& col("unit_price").isNotNull() & (col("unit_price") >= 0)
& col("order_date").isNotNull()
)
clean = valid.dropDuplicates()
return clean
clean = clean_orders("/home/labex/project/data/dirty-orders.csv")
clean.write.mode("overwrite").parquet("/home/labex/project/output/clean-orders")
reloaded = spark.read.parquet("/home/labex/project/output/clean-orders")
print("Read-back clean rows:")
reloaded.orderBy("order_id").show(30, truncate=False)
reloaded.printSchema()
print("Read-back count:", reloaded.count())
practice = clean_orders("/home/labex/project/data/dirty-orders-extra.csv")
practice.write.mode("overwrite").parquet("/home/labex/project/output/practice-clean-orders")
practice_reloaded = spark.read.parquet("/home/labex/project/output/practice-clean-orders")
print("Extra-delivery clean rows:")
practice_reloaded.orderBy("order_id").show(30, truncate=False)
print("Extra-delivery count:", practice_reloaded.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/clean_orders.py > /home/labex/project/output/clean-parquet-check.txt
cat /home/labex/project/output/clean-parquet-check.txt
Ambas as tabelas lidas de volta deverão conter o1 a o7 exatamente uma vez, e ambas as contagens deverão ser 7. O diretório principal de saída é output/clean-orders; practice-clean-orders é um experimento derivado separado. O esquema lido de volta deverá manter quantity inteiro, preço decimal e order_date como data.
Este é um ciclo mínimo de gravação e leitura de Parquet, não uma comparação de velocidade nem um exercício de ajuste de armazenamento. Agora você tem um pequeno conjunto limpo que outro processo Spark pode ler e evidências de que uma linha inválida e uma duplicata recém-introduzidas não alteram silenciosamente o resultado.
Resumo
Você preservou texto bruto, usou conversão segura para expor valores inválidos, aplicou uma política explícita de valores ausentes e intervalos e removeu linhas duplicadas idênticas. Manteve preços zero válidos e registros não pagos e verificou a saída real tipada em Parquet e um segundo recebimento. O próximo desafio permite aplicar essas habilidades de limpeza e cálculo de valores de forma independente.
