Ingerir dados com notebooks

Os notebooks no Azure Databricks fornecem uma abordagem flexível e orientada por código para ingerir dados de várias fontes. Quando as ferramentas gráficas não atendem aos seus requisitos ou quando você precisa de lógica personalizada para a transformação de dados, os notebooks fornecem controle total sobre o processo de ingestão. Você pode usar Python (PySpark), SQL, Scala ou R para se conectar a fontes de dados, aplicar transformações e carregar dados em tabelas do Catálogo do Unity.

Nesta unidade, você aprenderá a ingerir dados usando notebooks para cenários de lote e streaming.

Ingerir dados em lote com DataFrames

DataFrames são a base para a ingestão de dados em lote no Databricks. Um DataFrame é uma coleção distribuída de dados organizados em colunas nomeadas, semelhante a uma tabela em um banco de dados relacional. Use a spark.read API para carregar dados de várias fontes em um DataFrame.

O exemplo a seguir lê um arquivo CSV em um 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 arquivo, enquanto option() os métodos configuram como os dados são analisados. A opção header indica ao Spark que use a primeira linha como nomes de coluna, e o inferSchema detecta automaticamente tipos de dados.

Ler de diferentes formatos de arquivo

O Azure Databricks dá suporte a vários formatos de arquivo. Você pode ajustar o formato e as opções com base na 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 arquivos Parquet armazenam dados em um formato columnar, o que melhora o desempenho da consulta para cargas de trabalho analíticas. Os arquivos JSON dão suporte a estruturas aninhadas e são comuns para respostas de API. Os arquivos XML exigem a opção rowTag para especificar o elemento a ser mapeado em cada linha.

Conectar-se a bancos de dados externos

Para dados armazenados em bancos de dados relacionais, use conexões JDBC para ler diretamente da origem:

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())

Armazene credenciais em segredos do Databricks em vez de inseri-las diretamente em notebooks. Você também pode enviar consultas por push para o banco 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()

Gravar dados em tabelas do Catálogo do Unity

Depois de carregar e transformar seus dados, escreva-os em uma tabela do Catálogo do Unity. O Azure Databricks usa o formato Delta Lake por padrão, que fornece transações ACID e controle de versão:

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

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

• sobrescrever: substitui totalmente o conteúdo da tabela
• acrescentar: adiciona linhas à tabela existente
• ignore: Não fará nada se a tabela já existir
• erro: gera uma exceção se a tabela existir (padrão)

Ingerir dados de streaming com streaming estruturado

O Streaming Estruturado permite que você processe dados continuamente conforme eles chegam. Em vez de spark.read, você usa spark.readStream para criar um DataFrame de streaming que processa novos dados incrementalmente.

O exemplo a seguir lê dados de streaming do 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 onde começar a ler ao iniciar uma nova consulta. Use latest apenas para novos dados ou earliest para ler desde o início do tópico. Depois de configurar um local de ponto de verificação no fluxo de gravação, as execuções subsequentes serão retomadas de forma automática do último deslocamento processado, ignorando startingOffsets.

Processar fluxos baseados em arquivo

Você também pode transmitir de fontes de arquivo à medida que novos arquivos chegam em 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 arquivo de streaming, você deve definir o esquema explicitamente em vez de usar inferSchema. Não há suporte para inferência de esquema para fontes de streaming porque requer a leitura de todo o conjunto de dados.

Gravar dados de streaming em tabelas

Use writeStream para persistir os dados de streaming nas tabelas do Catálogo do 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 registros são gravados:

• acrescentar: somente novas linhas são gravadas (padrão para a maioria das operações)
• atualizar: as linhas alteradas são gravadas
• completo: todas as linhas são reescritas (utilizado com agregações)

O local do ponto de verificação armazena informações de progresso, permitindo que o fluxo seja retomado de onde parou, caso interrompido.

Manipular dados semiestruturados

Ao trabalhar com dados semiestruturados, como JSON, o tipo de dados VARIANT fornece 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"))

Em seguida, você pode 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()

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