Ingerir datos con notebooks

Los cuadernos de Azure Databricks proporcionan un enfoque flexible basado en código para ingerir datos de varios orígenes. Cuando las herramientas gráficas no cumplen sus requisitos o cuando se necesita lógica personalizada para la transformación de datos, los cuadernos le proporcionan un control completo sobre el proceso de ingesta. Puede usar Python (PySpark), SQL, Scala o R para conectarse a orígenes de datos, aplicar transformaciones y cargar datos en tablas de Catálogo de Unity.

En esta unidad, aprenderá cómo ingerir datos mediante cuadernos tanto para escenarios de procesamiento por lotes como de transmisión.

Ingesta de datos por lotes con DataFrames

DataFrames son la base para la ingesta de datos por lotes en Databricks. DataFrame es una colección distribuida de datos organizados en columnas con nombre, similar a una tabla de una base de datos relacional. Utilizas la API spark.read para cargar datos de varios orígenes en un DataFrame.

En el ejemplo siguiente se lee un archivo CSV en un dataframe:

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

El format() método especifica el tipo de archivo, mientras que option() los métodos configuran cómo se analizan los datos. La header opción indica a Spark que use la primera fila como nombres de columna y inferSchema detecte automáticamente los tipos de datos.

Lectura de diferentes formatos de archivo

Azure Databricks admite varios formatos de archivo. Puede ajustar el formato y las opciones en función del origen de datos:

# 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>

Los archivos Parquet almacenan datos en un formato de columnas, lo que mejora el rendimiento de las consultas para las cargas de trabajo analíticas. Los archivos JSON admiten estructuras anidadas y son comunes para las respuestas de API. Los archivos XML requieren la rowTag opción de especificar qué elemento se asigna a cada fila.

Conexión a bases de datos externas

Para los datos almacenados en bases de datos relacionales, use conexiones JDBC para leer directamente desde el origen:

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

Almacene las credenciales en secretos de Databricks en lugar de codificarlas de forma dura en cuadernos. También puede insertar consultas en la base de datos mediante una subconsulta en la dbtable opción :

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

Escritura de datos en tablas del Catálogo de Unity

Después de cargar y transformar los datos, escríbalos en una tabla de Catálogo de Unity. Azure Databricks usa el formato Delta Lake de forma predeterminada, que proporciona transacciones ACID y control de versiones:

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

La mode() opción controla cómo se controlan los datos existentes:

• sobrescribir: sustituye por completo el contenido de la tabla.
• append: agrega filas a la tabla existente
• ignore: no hace nada si la tabla ya existe
• error: produce una excepción si la tabla existe (valor predeterminado)

Ingesta de datos de streaming con Structured Streaming

Structured Streaming le permite procesar datos continuamente a medida que llega. En lugar de spark.read, se usa spark.readStream para crear un DataFrame de streaming que procese los nuevos datos de forma incremental.

En el ejemplo siguiente se leen datos de streaming de Kafka:

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

La startingOffsets opción determina dónde empezar a leer al iniciar una nueva consulta. Use latest solo para nuevos datos o earliest para leer desde el principio del tema. Una vez configurada una ubicación de punto de control en el flujo de escritura, las ejecuciones posteriores se reanudan automáticamente desde el último desplazamiento procesado, ignorando startingOffsets.

Procesar secuencias basadas en archivos

También puede transmitir desde fuentes de archivos cuando llegan nuevos archivos a un directorio.

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

Para los orígenes de archivos de streaming, debe definir el esquema de forma explícita en lugar de usar inferSchema. No se admite la inferencia de esquemas para los orígenes de streaming porque requiere leer todo el conjunto de datos.

Escritura de datos de streaming en tablas

Use writeStream para conservar los datos de streaming en tablas de Catálogo de Unity:

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

El modo de salida determina qué registros se escriben:

• append: solo se escriben nuevas filas (valor predeterminado para la mayoría de las operaciones)
• actualización: Se escriben las filas modificadas
• complete: todas las filas se reescriben (se utiliza con agregaciones)

La ubicación del punto de control almacena información de progreso, lo que permite que la secuencia se reanude desde donde se detuvo si se interrumpe.

Control de datos semiestructurados

Al trabajar con datos semiestructurados como JSON, el tipo de datos VARIANT proporciona flexibilidad para almacenar y consultar estructuras anidadas:

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

A continuación, puede consultar campos específicos mediante la variant_get función o la sintaxis SQL.

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

Este enfoque funciona bien cuando el esquema JSON varía entre registros o evoluciona con el tiempo.