Introdução
Uma empresa varejista recebe registros de pedidos e quer transformá-los em informações úteis. Antes de calcular relatórios de vendas, você precisa de uma forma de representar linhas, descrever suas colunas e pedir à ferramenta de processamento que execute trabalho. Você começará com três registros pequenos o suficiente para conferir visualmente.
Big data descreve cargas de trabalho de dados cujo tamanho, ritmo de chegada ou necessidades de processamento podem ultrapassar uma abordagem conveniente em uma única máquina. O Apache Spark é um mecanismo de processamento de dados que pode dividir trabalho entre máquinas em um cluster. O PySpark é sua interface Python. Um DataFrame representa linhas com colunas nomeadas e tipadas e aceita operações como filtragem e agregação. O Spark processa dados; um arquivo ou sistema de armazenamento separado fornece entrada e saída persistentes. Este curso começa com exemplos minúsculos para que você avalie a correção antes da escala.
Você criará uma SparkSession, exibirá um DataFrame, inspecionará seu esquema e sua contagem de linhas e adicionará um registro antes de executar novamente. O ambiente preparado usa local[2]: o Spark é executado nesta única máquina com duas threads de execução. Isso ensina APIs reais do Spark, enquanto a implantação distribuída é um tópico separado. Você precisa conhecer variáveis, listas e chamadas de funções básicas do Python; a trilha de Python pode ajudar com esses pré-requisitos. Não é necessário pandas, Hadoop, Hive, Java, instalação nem arquivos de laboratórios anteriores. Use o Terminal.
Execute um script completo de um pequeno DataFrame
Nesta etapa, você executará um script completo que transforma três registros Python em uma tabela Spark. Comece no espaço de trabalho:
cd /home/labex/project
Uma SparkSession é o ponto de entrada para criar DataFrames e acessar as APIs de dados estruturados do Spark. O builder define um nome de aplicação, seleciona execução local e cria a sessão. local[2] permite duas threads de execução nesta máquina; não cria duas máquinas de cluster. setLogLevel reduz os logs informativos posteriores para facilitar encontrar a saída didática.
No fonte Python, from pyspark.sql import SparkSession disponibiliza essa classe instalada ao programa. Um ponto acessa um membro nomeado, e parênteses chamam um método: df.show() solicita que este DataFrame exiba suas linhas. O builder encadeia várias chamadas; os parênteses ao redor permitem que essa expressão ocupe várias linhas. Você pode ler cada linha como uma configuração antes de getOrCreate() obter a sessão.
A lista orders contém três tuplas, grupos de valores entre parênteses. Cada tupla fornece uma linha na mesma ordem das colunas declaradas. A string de esquema nomeia cada coluna e seu tipo: identificadores e produtos como texto, seguidos de uma quantidade inteira. createDataFrame usa esses registros e esse esquema. show pede ao Spark que exiba linhas, e stop libera a sessão ao terminar.
Salve o script completo. O > do shell grava o arquivo nomeado; o marcador PYTHON entre aspas inicia um here-document que preserva literalmente o texto Python contido. Copie até a linha final PYTHON:
cat > /home/labex/project/first_dataframe.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = (
SparkSession.builder
.appName("FirstOrderTable")
.master("local[2]")
.getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")
orders = [
("s1", "notebook", 2),
("s2", "pen", 1),
("s3", "folder", 3),
]
columns = "order_id string, product string, quantity int"
df = spark.createDataFrame(orders, schema=columns)
df.show()
spark.stop()
PYTHON
O comando preparado bigdata-python seleciona o ambiente Python com PySpark instalado. Execute o arquivo como um programa Python comum, salvando sua saída normal localmente:
bigdata-python /home/labex/project/first_dataframe.py > /home/labex/project/output/first-run.txt
cat /home/labex/project/output/first-run.txt
Espere uma tabela com as colunas order_id, product e quantity, contendo s1/notebook/2, s2/pen/1 e s3/folder/3. Avisos de inicialização ou mensagens de progresso ainda podem aparecer no terminal; eles são separados da tabela salva. Nenhuma ordem de linhas é garantida sem uma ordenação explícita, portanto identifique os registros por seus IDs de pedido.
Seu programa Python é o driver: ele cria a sessão e pede ao Spark que realize trabalho. Em um cluster, processos executores executam tarefas sobre partes dos dados. Aqui, a execução acontece localmente na mesma máquina. Você tem um programa funcional sem precisar configurar um cluster.
Inspecione o esquema e conte as linhas
Nesta etapa, você distinguirá o que uma coluna pode conter da quantidade de registros da tabela. Um esquema lista nomes de colunas, tipos de dados e possibilidade de valores nulos. Uma coluna inteira representa números inteiros adequados a cálculos de quantidade; texto tem operações diferentes. Nullable significa que o campo permite um valor ausente, não que ele esteja ausente em todas as linhas.
Adicione printSchema para inspecionar essa descrição e count para calcular a quantidade de linhas. Aqui está o arquivo atualizado completo para que não haja estado interativo oculto:
cat > /home/labex/project/first_dataframe.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = (
SparkSession.builder
.appName("FirstOrderTable")
.master("local[2]")
.getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")
orders = [
("s1", "notebook", 2),
("s2", "pen", 1),
("s3", "folder", 3),
]
columns = "order_id string, product string, quantity int"
df = spark.createDataFrame(orders, schema=columns)
df.show()
df.printSchema()
print("Row count:", df.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/first_dataframe.py > /home/labex/project/output/schema-and-count.txt
cat /home/labex/project/output/schema-and-count.txt
A tabela ainda deverá conter três registros. Abaixo dela, procure:
root
|-- order_id: string (nullable = true)
|-- product: string (nullable = true)
|-- quantity: integer (nullable = true)
Row count: 3
printSchema descreve o DataFrame. show e count são ações que pedem ao Spark um resultado ou valor visível. O Spark pode adiar grande parte do processamento até que uma ação solicite uma resposta; laboratórios posteriores explicarão essa execução preguiçosa com transformações e planos. Uma contagem de 3 significa três registros de pedidos, não a soma das quantidades, que responderia a outra pergunta de negócio.
Cada chamada inicia um novo processo Python e cria sua própria SparkSession. O arquivo cria df explicitamente; ele não depende de um shell interativo ter criado spark ou df antes.
Preveja e teste um registro adicional
Nesta etapa, você mudará os dados mantendo sua estrutura. Preveja o que adicionar s4/notebook/1 muda: a quantidade de linhas de pedidos passa a quatro, o esquema permanece igual e os três registros originais continuam presentes. A nova linha é um pedido, embora notebook já apareça em outro pedido.
Substitua o script por esta versão, que adiciona apenas esse registro:
cat > /home/labex/project/first_dataframe.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = (
SparkSession.builder
.appName("FirstOrderTable")
.master("local[2]")
.getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")
orders = [
("s1", "notebook", 2),
("s2", "pen", 1),
("s3", "folder", 3),
("s4", "notebook", 1),
]
columns = "order_id string, product string, quantity int"
df = spark.createDataFrame(orders, schema=columns)
df.show()
df.printSchema()
print("Row count:", df.count())
spark.stop()
PYTHON
Execute o arquivo novamente em um novo processo:
bigdata-python /home/labex/project/first_dataframe.py > /home/labex/project/output/expanded-run.txt
cat /home/labex/project/output/expanded-run.txt
Localize os quatro IDs de pedido e Row count: 4. O esquema ainda deverá descrever os mesmos três nomes e tipos de coluna. Repetir um nome de produto não agrupa pedidos nem remove duplicatas automaticamente; essas operações devem ser solicitadas explicitamente em lições posteriores.
Os arquivos anteriores first-run.txt e schema-and-count.txt permanecem como evidências da versão de três linhas. O arquivo executável agora descreve a versão de quatro linhas, portanto outra execução independente pode reproduzir esse resultado. Você concluiu um pequeno ciclo de alterar, prever, executar e conferir, que ajudará a avaliar fluxos maiores de dados em vez de simplesmente confiar que um programa terminou com sucesso.
Resumo
Você executou um script PySpark real com sua própria SparkSession, criou um DataFrame tipado e conferiu seus registros exibidos, esquema e contagem de linhas. Adicionar um registro alterou a contagem de linhas, preservando o esquema. A seguir, passará de registros escritos dentro do Python para um arquivo CSV com esquema explícito.
