Introdução
Alguns sistemas entregam registros como linhas CSV; outros entregam objetos JSON com campos nomeados. A empresa varejista agora fornece arquivos de eventos contendo um rótulo de tipo e valores relacionados a pedidos. Suas perguntas de processamento são conhecidas: o que os campos significam, quais registros pertencem a pedidos e quais colunas o resultado deve expor?
JSON Lines armazena um objeto JSON completo em cada linha. Diferentemente do CSV, os nomes dos campos acompanham cada objeto, e um objeto pode conter outro. Você primeiro lerá registros planos com tipos explícitos, selecionará campos de um objeto simples de cliente aninhado e aplicará esse esquema a um segundo recebimento.
Estes são arquivos finitos lidos com spark.read, portanto isto ainda é processamento em lote. Um nome de arquivo contendo events não torna uma aplicação streaming. O ambiente e os dados de exemplo são independentes; você precisa dos conceitos anteriores de esquema explícito, select, filter e alias, além de Python básico. Não é necessário Kafka, Hadoop, Hive, configuração de streaming, arrays nem recursão complexa. Use o Terminal.
Leia JSON Lines plano e selecione eventos de pedidos
Nesta etapa, você transferirá a leitura com esquema explícito do CSV para o JSON. Inspecione o arquivo fornecido:
cd /home/labex/project
cat /home/labex/project/data/flat-events.jsonl
Cada linha é um objeto delimitado por chaves. Nomes entre aspas duplas identificam campos; dois-pontos separam um nome de seu valor. Números como quantity são valores numéricos JSON, enquanto order_date é uma string de data entre aspas que o DateType declarado interpretará. Não há linha de cabeçalho CSV nem um array ao redor.
Os campos order_id, quantidade, preço e data do primeiro objeto descrevem o mesmo tipo de valores de um pedido CSV. O arquivo também contém um evento de inventário; ler um registro válido não significa que ele pertença a um resultado de pedidos. O esquema explícito abaixo declara os campos planos. json(path) lê este formato, enquanto o filter conhecido seleciona event_type == order.
Salve o script completo com um here-document entre aspas e execute-o pelo interpretador PySpark preparado:
cat > /home/labex/project/read_json_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("ReadJSONEvents").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
event_schema = StructType([
StructField("event_type", StringType(), True),
StructField("order_id", StringType(), True),
StructField("quantity", IntegerType(), True),
StructField("unit_price", DecimalType(10, 2), True),
StructField("order_date", DateType(), True),
])
events = (
spark.read.schema(event_schema)
.option("dateFormat", "yyyy-MM-dd")
.json("/home/labex/project/data/flat-events.jsonl")
)
events.printSchema()
events.orderBy("order_id").show(truncate=False)
orders = events.filter(col("event_type") == "order")
print("Order events:")
orders.select("order_id", "quantity", "unit_price", "order_date").orderBy("order_id").show(truncate=False)
spark.stop()
PYTHON
bigdata-python /home/labex/project/read_json_orders.py > /home/labex/project/output/flat-json-orders.txt
cat /home/labex/project/output/flat-json-orders.txt
A exibição inicial deverá incluir os três registros de eventos. Sob Order events, espere apenas j1/2/10.00/2026-10-14 e j2/1/25.00/2026-10-14. O registro de inventário foi lido com sucesso, mas excluído pela condição de tipo. O esquema deverá mostrar quantity inteiro, preço decimal e order_date como data.
O leitor usa os campos JSON nomeados com os tipos declarados; diferentemente de um registro CSV simples, a ordem dos campos dentro de cada objeto JSON não define a identidade da coluna. O esquema ainda define o contrato pretendido.
Selecione campos de um objeto de cliente aninhado
Nesta etapa, você lerá um nível estrutural adicional. Inspecione o arquivo de eventos de clientes:
cat /home/labex/project/data/customer-events.jsonl
O valor customer é outro objeto, contendo customer_id e region. No esquema do Spark, isso se torna um struct, um campo que contém campos filhos nomeados. Não é um array nem uma segunda tabela para fazer uma junção.
Declare esse struct com um StructType interno. col("customer.region") seleciona o campo filho pelo caminho com pontos; alias("region") dá ao resultado selecionado um nome simples de coluna. Aplique a mesma ideia ao identificador de cliente:
cat > /home/labex/project/read_json_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("ReadJSONEvents").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
event_schema = StructType([
StructField("event_type", StringType(), True),
StructField("order_id", StringType(), True),
StructField("customer", StructType([
StructField("customer_id", StringType(), True),
StructField("region", StringType(), True),
]), True),
StructField("quantity", IntegerType(), True),
StructField("unit_price", DecimalType(10, 2), True),
StructField("order_date", DateType(), True),
])
events = (
spark.read.schema(event_schema)
.option("dateFormat", "yyyy-MM-dd")
.json("/home/labex/project/data/customer-events.jsonl")
)
events.printSchema()
orders = events.filter(col("event_type") == "order")
selected = orders.select(
"order_id",
col("customer.customer_id").alias("customer_id"),
col("customer.region").alias("region"),
"quantity", "unit_price", "order_date",
)
print("Selected customer orders:")
selected.orderBy("order_id").show(truncate=False)
selected.printSchema()
spark.stop()
PYTHON
bigdata-python /home/labex/project/read_json_orders.py > /home/labex/project/output/nested-json-orders.txt
cat /home/labex/project/output/nested-json-orders.txt
O esquema completo deverá mostrar customer como um struct com dois filhos de string. A tabela selecionada deverá ter j3/c1/East/2/10.00/2026-10-15 e j4/c2/West/1/25.00/2026-10-15. Seu esquema final expõe customer_id e region como colunas comuns de string ao lado de quantidade, preço e data.
Você selecionou valores já presentes em cada evento de pedido. Nenhuma tabela de consulta nem junção foi necessária. A representação aninhada original continua disponível em events, enquanto selected oferece uma visão plana mais simples para as próximas operações.
Reutilize o esquema aninhado em outro arquivo
Nesta etapa, você mudará o recebimento preservando a mesma estrutura. Primeiro, leia o novo arquivo:
cat /home/labex/project/data/new-customer-events.jsonl
Preveja a saída: um pedido j5 de c2 em West, quantidade 3 e preço 12.50 em 16 de outubro. O leitor e a seleção aninhada podem permanecer iguais; apenas o caminho de entrada muda. Execute esta versão completa:
cat > /home/labex/project/read_json_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("ReadJSONEvents").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
event_schema = StructType([
StructField("event_type", StringType(), True),
StructField("order_id", StringType(), True),
StructField("customer", StructType([
StructField("customer_id", StringType(), True),
StructField("region", StringType(), True),
]), True),
StructField("quantity", IntegerType(), True),
StructField("unit_price", DecimalType(10, 2), True),
StructField("order_date", DateType(), True),
])
events = (
spark.read.schema(event_schema)
.option("dateFormat", "yyyy-MM-dd")
.json("/home/labex/project/data/new-customer-events.jsonl")
)
events.printSchema()
orders = events.filter(col("event_type") == "order")
selected = orders.select(
"order_id",
col("customer.customer_id").alias("customer_id"),
col("customer.region").alias("region"),
"quantity", "unit_price", "order_date",
)
print("Selected customer orders:")
selected.orderBy("order_id").show(truncate=False)
selected.printSchema()
spark.stop()
PYTHON
bigdata-python /home/labex/project/read_json_orders.py > /home/labex/project/output/new-json-orders.txt
cat /home/labex/project/output/new-json-orders.txt
Espere o resultado previsto de uma linha e os mesmos tipos selecionados. Um novo arquivo não exige inventar um novo esquema quando a estrutura acordada não mudou.
As três leituras usaram spark.read sobre arquivos existentes e terminaram quando seus scripts foram concluídos. Elas não se inscreveram para receber mensagens futuras. Você aprendeu a variar um formato de entrada sem mudar simultaneamente o modelo de execução. A seguir, voltará a um recebimento propositalmente imperfeito de pedidos e definirá regras explícitas de limpeza.
Resumo
Você leu JSON Lines estático com tipos escalares explícitos, selecionou eventos de pedidos e transformou um struct simples de cliente em colunas usando caminhos nomeados. Um segundo recebimento usou corretamente o mesmo esquema. A seguir, tratará registros de pedidos ausentes, duplicados e inválidos sob regras claras de limpeza.
