Ingesta incremental y de streaming

Azure Databricks usa Apache Spark Structured Streaming para respaldar numerosos productos asociados a cargas de trabajo de ingesta, entre las que se incluyen:

• Cargador automático
• COPY INTO
• Canalizaciones de Delta Live Tables
• Vistas materializadas y tablas de streaming de Databricks SQL

En este artículo se describen algunas de las diferencias entre la semántica de procesamiento por lotes incremental y streaming y se proporciona información general de alto nivel sobre la configuración de cargas de trabajo de ingesta para la semántica deseada en Databricks.

¿Cuál es la diferencia entre la ingesta por lotes incremental y streaming?

Las posibles configuraciones de flujo de trabajo de ingesta van desde el procesamiento casi en tiempo real hasta el procesamiento por lotes incremental poco frecuente. Ambos patrones usan Apache Spark Structured Streaming para impulsar el procesamiento incremental, pero tienen una semántica diferente. Para simplificar, en este artículo se hace referencia a la ingesta casi en tiempo real como ingesta de streaming y un procesamiento incremental poco frecuente como ingesta incremental.

Ingesta de streaming

El streaming, en el contexto de las actualizaciones de ingesta y tabla de datos, hace referencia al procesamiento de datos casi en tiempo real en el que Azure Databricks ingiere registros de origen a receptor en microbachas mediante la infraestructura always-on. Una carga de trabajo de streaming ingiere continuamente actualizaciones de orígenes de datos configurados, a menos que se produzca un error que detenga la ingesta.

Ingesta por lotes incremental

La ingesta por lotes incremental hace referencia a un patrón en el que todos los registros nuevos se procesan desde un origen de datos en un trabajo de corta duración. La ingesta incremental por lotes suele producirse según una programación, pero también se puede desencadenar manualmente o en función de la llegada de archivos.

La ingesta incremental por lotes difiere de la ingesta por lotes en que detecta automáticamente nuevos registros en el origen de datos y omite los registros que ya se han ingerido.

Ingesta con trabajos

Los trabajos de Databricks permiten organizar flujos de trabajo y programar tareas que incluyen cuadernos, bibliotecas, canalizaciones de Delta Live Tables y consultas SQL de Databricks.

Nota:

Puede usar todos los tipos de proceso y tipos de tareas de Azure Databricks para configurar la ingesta incremental por lotes. La ingesta de streaming solo se admite en producción en el proceso de trabajos clásicos y Delta Live Tables.

Los Trabajos tienen dos modos principales de funcionamiento:

• Los trabajos continuos vuelven a intentarlo automáticamente si se produce un error. Este modo está pensado para la ingesta de streaming.
• Los trabajos desencadenados ejecutan tareas cuando se desencadenan. Entre los desencadenadores se incluyen:
  • Desencadenadores basados en tiempo que ejecutan trabajos según una programación especificada.
  • Desencadenadores basados en archivos que ejecutan trabajos cuando los archivos llegan a una ubicación especificada.
  • Otros desencadenadores, como las llamadas a la API REST, la ejecución de comandos de la CLI de Azure Databricks o el botón Ejecutar ahora en la interfaz de usuario del área de trabajo.

En el caso de las cargas de trabajo por lotes incrementales, configure los trabajos mediante el AvailableNow modo de desencadenador, como se indica a continuación:

Python

(df.writeStream
  .option("checkpointLocation", <checkpoint-path>)
  .trigger(availableNow=True)
  .toTable("table_name")
)

Scala

import org.apache.spark.sql.streaming.Trigger

df.writeStream
  .option("checkpointLocation", <checkpoint-path>)
  .trigger(Trigger.AvailableNow)
  .toTable("table_name")

En el caso de las cargas de trabajo de streaming, el intervalo de desencadenador predeterminado es processingTime ="500ms". En el ejemplo siguiente se muestra cómo procesar un microproceso cada 5 segundos:

Python

(df.writeStream
  .option("checkpointLocation", <checkpoint-path>)
  .trigger(processingTime="5 seconds")
  .toTable("table_name")
)

Scala

import org.apache.spark.sql.streaming.Trigger

df.writeStream
  .option("checkpointLocation", <checkpoint-path>)
  .trigger(Trigger.ProcessingTime, "5 seconds")
  .toTable("table_name")

Importante

Los trabajos sin servidor no admiten intervalos de desencadenador basados en Scala, modo continuo o basado en tiempo para Structured Streaming. Use trabajos clásicos si necesita una semántica de ingesta casi en tiempo real.

Ingesta con tablas dinámicas Delta

De forma similar a Trabajos, las canalizaciones de Delta Live Tables se pueden ejecutar en modo desencadenado o continuo. Para la semántica de streaming casi en tiempo real con tablas de streaming, use el modo continuo.

Use tablas de streaming para configurar la ingesta por lotes o streaming incremental desde el almacenamiento de objetos en la nube, Apache Kafka, Amazon Kinesis, Google Pub/Sub o Apache Pulsar.

LakeFlow Connect usa Delta Live Tables para configurar canalizaciones de ingesta desde sistemas conectados. Consulte LakeFlow Connect.

Las vistas materializadas garantizan la semántica de las operaciones equivalentes a las cargas de trabajo por lotes, pero pueden optimizar muchas operaciones para calcular los resultados de forma incremental. Consulte Operaciones de actualización para vistas materializadas.

Ingesta con Databricks SQL

Puede usar tablas de streaming para configurar la ingesta por lotes incremental desde el almacenamiento de objetos en la nube, Apache Kafka, Amazon Kinesis, Google Pub/Sub o Apache Pulsar.

Puede usar vistas materializadas para configurar el procesamiento por lotes incremental desde orígenes que son totalmente reproducibles para un conjunto de operaciones especificado. Consulte Operaciones de actualización para vistas materializadas.

COPY INTO proporciona una sintaxis SQL conocida para el procesamiento por lotes incremental para archivos de datos en el almacenamiento de objetos en la nube. el comportamiento COPY INTO es similar a los patrones admitidos por las tablas de streaming para el almacenamiento de objetos en la nube, pero no todas las configuraciones predeterminadas son equivalentes para todos los formatos de archivo admitidos.