Ingerir dados com notebooks

Os notebooks no Azure Databricks oferecem uma abordagem flexível e orientada por código para a absorção de dados de várias fontes. Quando as ferramentas gráficas não cumprem os seus requisitos, ou quando precisa de lógica personalizada para transformação de dados, os cadernos dão-lhe controlo total sobre o processo de ingestão. Pode usar Python (PySpark), SQL, Scala ou R para ligar a fontes de dados, aplicar transformações e carregar dados em tabelas do Catálogo Unity.

Nesta unidade, aprende a ingerir dados usando notebooks tanto para cenários em lote como para streaming.

Ingerir dados em lote com DataFrames

Os DataFrames são a base para a ingestão de dados em lote em Databricks. Um DataFrame é uma coleção distribuída de dados organizada em colunas nomeadas, semelhante a uma tabela numa base de dados relacional. Usas a spark.read API para carregar dados de várias fontes num DataFrame.

O exemplo seguinte lê um ficheiro CSV num DataFrame:

df = (spark.read
    .format("csv")
    .option("header", "true")
    .option("inferSchema", "true")
    .load("/Volumes/catalog/schema/volume/data.csv"))

O format() método especifica o tipo de ficheiro, enquanto option() os métodos definem como os dados são analisados. A opção header indica ao Spark para usar a primeira linha como nomes das colunas, e inferSchema deteta automaticamente os tipos de dados.

Ler de diferentes formatos de ficheiro

O Azure Databricks suporta múltiplos formatos de ficheiro. Pode ajustar o formato e as opções com base na sua fonte de dados:

# Read JSON files
df_json = spark.read.format("json").load("/path/to/data.json")

# Read Parquet files (columnar format, efficient for analytics)
df_parquet = spark.read.format("parquet").load("/path/to/data.parquet")

# Read XML files (requires rowTag option)
df_xml = (spark.read
    .format("xml")
    .option("rowTag", "record")
    .load("/path/to/data.xml"))

<data>
  <record><id>1</id><name>Alice</name></record>
  <record><id>2</id><name>Bob</name></record>
</data>

Os ficheiros Parquet armazenam dados num formato colunar, o que melhora o desempenho das consultas para cargas de trabalho analíticas. Os ficheiros JSON suportam estruturas aninhadas e são comuns para respostas de API. Os ficheiros XML requerem a rowTag opção de especificar qual elemento corresponde a cada linha.

Liga-te a bases de dados externas

Para dados armazenados em bases de dados relacionais, use ligações JDBC para ler diretamente da fonte:

df_jdbc = (spark.read
    .format("jdbc")
    .option("url", "jdbc:sqlserver://server.database.windows.net:1433;database=mydb")
    .option("dbtable", "schema.tablename")
    .option("user", dbutils.secrets.get(scope="jdbc", key="username"))
    .option("password", dbutils.secrets.get(scope="jdbc", key="password"))
    .load())

Armazena as credenciais nos segredos do Databricks em vez de as codificar diretamente em cadernos. Pode também enviar consultas para a base de dados usando uma subconsulta na opção dbtable.

pushdown_query = "(SELECT id, name, amount FROM orders WHERE order_date > '2024-01-01') AS filtered_orders"
df = spark.read.format("jdbc").option("dbtable", pushdown_query).load()

Escrever dados em tabelas do Unity Catalog

Depois de carregar e transformar os seus dados, escreva-os numa tabela do Unity Catalog. O Azure Databricks utiliza o formato Delta Lake por padrão, que fornece transações ACID e versionamento.

df.write.mode("overwrite").saveAsTable("catalog.schema.table_name")

A mode() opção controla como os dados existentes são tratados:

• sobrescrever: Substitui completamente o conteúdo da tabela
• append: Adiciona linhas à tabela existente
• ignorar: Não faz nada se a tabela já existir
• erro: Lança uma exceção se a tabela existir (padrão)

Ingerir dados de streaming com Streaming Estruturado

O Structured Streaming permite-lhe processar dados continuamente à medida que chegam. Em vez de spark.read, usa spark.readStream para criar um DataFrame em streaming que processa novos dados de forma incremental.

O exemplo seguinte lê dados em streaming de Kafka:

df_stream = (spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "broker:9092")
    .option("subscribe", "topic-name")
    .option("startingOffsets", "latest")
    .load())

A startingOffsets opção determina por onde começar a ler ao iniciar uma nova consulta. Use latest apenas para dados novos ou earliest para ler desde o início do tema. Depois de configurar um ponto de verificação no fluxo de escrita, as execuções subsequentes retomam automaticamente a partir do último deslocamento processado, ignorando startingOffsets.

Fluxos baseados em ficheiros de processo

Também pode transmitir a partir das fontes de ficheiros à medida que novos ficheiros chegam a um diretório:

df_file_stream = (spark.readStream
    .format("csv")
    .option("header", "true")
    .schema(defined_schema)
    .load("/Volumes/catalog/schema/volume/incoming/"))

Para fontes de ficheiros de streaming, deve definir explicitamente o esquema em vez de usar inferSchema. A inferência de esquema não é suportada para fontes de streaming porque requer ler todo o conjunto de dados.

Escrever dados em fluxo para tabelas

Utilize writeStream para persistir os dados em fluxo contínuo nas tabelas do Catálogo Unity.

(df_stream
    .writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "/checkpoints/my-stream")
    .toTable("catalog.schema.streaming_table"))

O modo de saída determina quais os registos que são escritos:

• append: Apenas novas linhas são escritas (padrão para a maioria das operações)
• atualização: As linhas alteradas estão escritas
• complete: Todas as linhas são reescritas (utilizadas com agregações)

A localização do ponto de controlo armazena informação de progresso, permitindo que o fluxo retome onde parou se for interrompido.

Lidar com dados semi-estruturados

Ao trabalhar com dados semi-estruturados como JSON, o tipo de dados VARIANT oferece flexibilidade para armazenar e consultar estruturas aninhadas:

from pyspark.sql.functions import parse_json, variant_get, col

df = spark.read.text("/path/to/data.json")
df_variant = df.select(parse_json(col("value")).alias("data"))

Pode então consultar campos específicos usando a variant_get função ou a sintaxe SQL:

df_variant.select(variant_get(col("data"), "$.customer.name", "string")).display()

Esta abordagem funciona bem quando o esquema JSON varia entre registos ou evolui ao longo do tempo.