Usar o Spark para trabalhar com arquivos de dados

Depois de configurar um bloco de anotações e anexá-lo a um cluster, você pode usar o Spark para ler e processar arquivos de dados. O Spark oferece suporte a uma ampla variedade de formatos, como CSV, JSON, Parquet, ORC, Avro e Delta, e o Databricks fornece conectores internos para acessar arquivos armazenados no espaço de trabalho, no Azure Data Lake ou no Armazenamento de Blobs ou em outros sistemas externos.

O fluxo de trabalho geralmente segue três etapas:

1. Leia um arquivo em um Spark DataFrame usando spark.read com o formato e o caminho corretos. Ao ler formatos de texto bruto, como CSV ou JSON, o Spark pode inferir o esquema (nomes de colunas e tipos de dados), mas isso às vezes é lento ou não confiável. Uma prática melhor na produção é definir o esquema explicitamente para que os dados sejam carregados de forma consistente e eficiente.

2. Explore e transforme o DataFrame usando operações SQL ou DataFrame (por exemplo, filtrando linhas, selecionando colunas, agregando valores).

3. Escreva os resultados de volta para o armazenamento em um formato escolhido.

O trabalho com arquivos no Spark foi projetado para ser consistente em conjuntos de dados pequenos e grandes. O mesmo código usado para testar um pequeno ficheiro CSV também funciona em conjuntos de dados muito maiores, uma vez que o Spark distribui o trabalho pelo cluster. Isso facilita a expansão da exploração rápida para o processamento de dados mais complexo.

Carregando dados em um dataframe

Considere um exemplo hipotético para ver como pode usar um dataframe para trabalhar com dados. Suponha que tem os seguintes dados num ficheiro de texto delimitado por vírgulas chamado products.csv armazenados num volume do Unity Catalog:

ProductID,ProductName,Category,ListPrice
771,"Mountain-100 Silver, 38",Mountain Bikes,3399.9900
772,"Mountain-100 Silver, 42",Mountain Bikes,3399.9900
773,"Mountain-100 Silver, 44",Mountain Bikes,3399.9900
...

Em um bloco de anotações do Spark, você pode usar o seguinte código PySpark para carregar os dados em um dataframe e exibir as primeiras 10 linhas:

%python
df = spark.read.load('/Volumes/my_catalog/my_schema/my_volume/products.csv',
    format='csv',
    header=True,
    inferSchema=True
)
display(df.limit(10))

A %python linha no início chama-se magic, e diz ao Spark que a linguagem usada nesta célula é Python. Aqui está o código Scala equivalente para o exemplo de dados de produtos:

%scala
val df = spark.read.format("csv").option("header", "true").option("inferSchema", "true").load("/Volumes/my_catalog/my_schema/my_volume/products.csv")
display(df.limit(10))

A magia %scala é usada para especificar Scala.

Ambos os exemplos mostrados anteriormente produziriam resultados como este:

Especificando um esquema de dataframe

No exemplo anterior, a primeira linha do ficheiro CSV continha os nomes das colunas, e o Spark inferiu o tipo de dado de cada coluna a partir dos dados que contêm porque o código definiu as opções header e inferSchema. Você também pode especificar um esquema explícito para os dados, que é útil quando os nomes das colunas não são incluídos no arquivo de dados, como este exemplo CSV:

771,"Mountain-100 Silver, 38",Mountain Bikes,3399.9900
772,"Mountain-100 Silver, 42",Mountain Bikes,3399.9900
773,"Mountain-100 Silver, 44",Mountain Bikes,3399.9900
...

O exemplo PySpark a seguir mostra como especificar um esquema para o dataframe a ser carregado de um arquivo chamado product-data.csv neste formato:

from pyspark.sql.types import *
from pyspark.sql.functions import *

productSchema = StructType([
    StructField("ProductID", IntegerType()),
    StructField("ProductName", StringType()),
    StructField("Category", StringType()),
    StructField("ListPrice", FloatType())
    ])

df = spark.read.load('/Volumes/my_catalog/my_schema/my_volume/product-data.csv',
    format='csv',
    schema=productSchema,
    header=False)
display(df.limit(10))

Os resultados seriam, mais uma vez, semelhantes a:

Filtragem e agrupamento de dataframes

Você pode usar os métodos da classe Dataframe para filtrar, classificar, agrupar e manipular os dados que ela contém. Por exemplo, o seguinte exemplo de código usa o select método para recuperar as colunas ProductID e ListPrice a partir do dataframe df contendo dados de produto no exemplo anterior:

pricelist_df = df.select("ProductID", "ListPrice")

Os resultados deste exemplo de código seriam mais ou menos assim:

Em comum com a maioria dos métodos de manipulação de dados, select retorna um novo objeto dataframe.

Você pode "encadear" métodos juntos para executar uma série de manipulações que resultam em um dataframe transformado. Por exemplo, este exemplo de código encadeia os select métodos e where para criar um novo dataframe contendo as colunas ProductName e ListPrice para produtos com uma categoria de Mountain Bikes ou Road Bikes:

bikes_df = df.select("ProductName", "ListPrice").where((df["Category"]=="Mountain Bikes") | (df["Category"]=="Road Bikes"))
display(bikes_df)

Os resultados deste exemplo de código seriam mais ou menos assim:

Para agrupar e agregar dados, você pode usar o método groupBy e funções de agregação. Por exemplo, o seguinte código PySpark conta o número de produtos para cada categoria:

counts_df = df.select("ProductID", "Category").groupBy("Category").count()
display(counts_df)

Os resultados deste exemplo de código seriam mais ou menos assim:

Usando expressões SQL no Spark

A API Dataframe faz parte de uma biblioteca do Spark chamada Spark SQL, que permite que os analistas de dados usem expressões SQL para consultar e manipular dados.

Criando objetos de banco de dados no catálogo do Spark

O catálogo do Spark é um metastore para objetos de dados relacionais, como exibições e tabelas. O runtime do Spark pode usar o catálogo para integrar código escrito em qualquer linguagem suportada pelo Spark com expressões SQL que possam ser mais naturais para alguns analistas de dados ou programadores.

Uma das maneiras mais simples de disponibilizar dados em um dataframe para consulta no catálogo do Spark é criar uma exibição temporária, conforme mostrado no exemplo de código a seguir:

df.createOrReplaceTempView("products")

Uma vista é temporária, o que significa que é automaticamente eliminada no final da sessão atual. Você também pode criar tabelas que são persistentes no catálogo para definir um banco de dados que pode ser consultado usando o Spark SQL.

Usando a API SQL do Spark para consultar dados

Você pode usar a API SQL do Spark em código escrito em qualquer idioma para consultar dados no catálogo. Por exemplo, o código PySpark a seguir usa uma consulta SQL para retornar dados da exibição de produtos como um dataframe.

bikes_df = spark.sql("SELECT ProductID, ProductName, ListPrice \
                      FROM products \
                      WHERE Category IN ('Mountain Bikes', 'Road Bikes')")
display(bikes_df)

Os resultados do exemplo de código seriam semelhantes à tabela a seguir:

Usando código SQL

O exemplo anterior demonstrou como usar a API SQL do Spark para incorporar expressões SQL no código do Spark. Num notebook, você também pode usar o comando mágico %sql para executar código SQL que consulta objetos no catálogo, como este:

%sql

SELECT Category, COUNT(ProductID) AS ProductCount
FROM products
GROUP BY Category
ORDER BY Category

O exemplo de código SQL retorna um conjunto de resultados que é exibido automaticamente no bloco de anotações como uma tabela, como a abaixo: