Compartir a través de

Carga de datos mediante tablas de streaming en Databricks SQL

  • Artículo

Databricks recomienda usar tablas de streaming para ingerir datos mediante Databricks SQL. Una tabla de streaming es una tabla registrada en Unity Catalog con compatibilidad adicional con el procesamiento de datos incremental o de transmisión. Se crea automáticamente una canalización de Delta Live Tables para cada tabla de streaming. Puede usar tablas de streaming para la carga incremental de datos desde Kafka y el almacenamiento de objetos en la nube.

En este artículo se muestra el uso de tablas de streaming para cargar datos desde el almacenamiento de objetos en la nube configurado como un volumen de Unity Catalog (recomendado) o una ubicación externa.

Nota:

Para obtener información sobre el modo de uso de las tablas de Delta Lake como orígenes y receptores de streaming, consulte Lecturas y escrituras de streaming de tablas de Delta.

Antes de empezar

Antes de comenzar, deberá cumplir los siguientes requisitos.

Requisitos del área de trabajo:

Requisitos de proceso:

Debe usar uno de los siguientes enfoques:

  • Un almacén de SQL que usa el canal Current.

  • Procesar con el modo de acceso compartido en Databricks Runtime 13.3 LTS o superior.

  • Proceso con modo de acceso de usuario único en Databricks Runtime 15.4 LTS o superior.

    En Databricks Runtime 15.3 y versiones posteriores, no puede usar un proceso de usuario único para consultar tablas de streaming que son propiedad de otros usuarios. Puede usar el proceso de usuario único en Databricks Runtime 15.3 y versiones posteriores solo si posee la tabla de streaming. El creador de la tabla es el propietario.

    Databricks Runtime 15.4 LTS y versiones posteriores admiten consultas en tablas generadas por Delta Live Tables en un solo proceso de usuario, independientemente de la propiedad de la tabla. Para aprovechar el filtrado de datos proporcionado en Databricks Runtime 15.4 LTS y versiones posteriores, debe confirmar que el área de trabajo está habilitada para el proceso sin servidor, ya que la funcionalidad de filtrado de datos que admite tablas generadas por Delta Live Tables se ejecuta en un proceso sin servidor. Se puede cobrar por los recursos de proceso sin servidor cuando se usa un proceso de usuario único para ejecutar operaciones de filtrado de datos. Consulte Control de acceso específico sobre el proceso de un solo usuario.

Requisitos de permisos:

Otros requisitos:

  • La ruta de acceso a los datos de origen.

    Ejemplo de ruta de acceso del volumen: /Volumes/<catalog>/<schema>/<volume>/<path>/<file-name>

    Ejemplo de ruta de acceso de ubicación externa: abfss://myContainer@myStorageAccount.dfs.core.windows.net/analysis

    Nota:

    En este artículo se da por supuesto que los datos que quiere cargar están en una ubicación de almacenamiento en la nube que se corresponde con un volumen de Unity Catalog o con una ubicación externa a la que tiene acceso.

Detección y vista previa de los datos de origen

  1. En la barra lateral del área de trabajo, haga clic en Consultas y, a continuación, en Crear consulta.

  2. En el editor de consultas, seleccione un almacén de SQL que use el canal Current en la lista desplegable.

  3. Pegue lo siguiente en el editor, sustituyendo los valores entre corchetes angulares (<>) por la información que identifica los datos de origen y, a continuación, haga clic en Ejecutar.

    Nota:

    Es posible que encuentre errores de inferencia de esquema al ejecutar la función con valores de tabla read_files si los valores predeterminados de la función no pueden analizar los datos. Por ejemplo, es posible que tenga que configurar el modo de varias líneas para archivos CSV o JSON de varias líneas. Para obtener una lista de las opciones del analizador, consulte función con valores de tabla read_files.

    /* Discover your data in a volume */
LIST "/Volumes/<catalog>/<schema>/<volume>/<path>/<folder>"

/* Preview your data in a volume */
SELECT * FROM read_files("/Volumes/<catalog>/<schema>/<volume>/<path>/<folder>") LIMIT 10

/* Discover your data in an external location */
LIST "abfss://<container>@<storage-account>.dfs.core.windows.net/<path>/<folder>"

/* Preview your data */
SELECT * FROM read_files("abfss://<container>@<storage-account>.dfs.core.windows.net/<path>/<folder>") LIMIT 10

Carga de datos en una tabla de streaming

Para crear una tabla de streaming a partir de datos en el almacenamiento de objetos en la nube, pegue lo siguiente en el editor de consultas y, a continuación, haga clic en Ejecutar:

/* Load data from a volume */
CREATE OR REFRESH STREAMING TABLE <table-name> AS
SELECT * FROM STREAM read_files('/Volumes/<catalog>/<schema>/<volume>/<path>/<folder>')

/* Load data from an external location */
CREATE OR REFRESH STREAMING TABLE <table-name> AS
SELECT * FROM STREAM read_files('abfss://<container>@<storage-account>.dfs.core.windows.net/<path>/<folder>')

Actualización de una tabla de streaming mediante una canalización DLT

En esta sección se describen los patrones para actualizar una tabla de streaming con los datos disponibles más recientes de los orígenes definidos en la consulta.

Cuando usted CREATE o REFRESH una tabla de streaming, la actualización procesa mediante una canalización de Delta Live Tables sin servidor. Cada tabla de streaming que defina tiene una canalización de Delta Live Tables asociada.

Después de ejecutar el comando REFRESH, se devuelve el vínculo de la canalización DLT. Puede usar el vínculo de la canalización DLT para comprobar el estado de la actualización.

Nota:

Solo el propietario de la tabla puede actualizar una tabla de streaming para obtener los datos más recientes. El usuario que crea la tabla es el propietario y no se puede cambiar el propietario. Es posible que tenga que actualizar la tabla de streaming antes de usar consultas de viaje de tiempo.

Consulte ¿Qué es Delta Live Tables?

Ingerir nuevos datos solo

De forma predeterminada, la función read_files lee todos los datos existentes en el directorio de origen durante la creación de la tabla y, a continuación, procesa los registros recién llegados con cada actualización.

Para evitar la ingesta de datos que ya existen en el directorio de origen en el momento de la creación de la tabla, establezca la opción includeExistingFiles en false. Así, solo se procesarán los datos que lleguen al directorio después de crear la tabla. Por ejemplo:

CREATE OR REFRESH STREAMING TABLE my_bronze_table
AS SELECT *
FROM STREAM read_files(
  'abfss://myContainer@myStorageAccount.dfs.core.windows.net/analysis/*/*/*.json',
  includeExistingFiles => false)

Actualización completa de una tabla de streaming

Las actualizaciones completas vuelven a procesar todos los datos disponibles en el origen con la definición más reciente. No se recomienda llamar a actualizaciones completas en orígenes que no mantengan todo el historial de los datos o tengan períodos de retención cortos, como Kafka, ya que la actualización completa trunca los datos existentes. Es posible que no pueda recuperar datos antiguos si los datos ya no están disponibles en el origen.

Por ejemplo:

REFRESH STREAMING TABLE my_bronze_table FULL

Programación de una tabla de streaming para la actualización automática

Para configurar una tabla de streaming de modo que se actualice automáticamente según una programación definida, pegue lo siguiente en el editor de consultas y, a continuación, haga clic en Ejecutar:

ALTER STREAMING TABLE
[[<catalog>.]<database>.]<name>
ADD [SCHEDULE [REFRESH]
        CRON '<cron-string>'
                [ AT TIME ZONE '<timezone-id>' ]];

Para ver ejemplos de consultas de programación de actualizaciones, consulte ALTER STREAMING TABLE.

Seguimiento del estado de una actualización

Para ver el estado de una actualización de tabla de streaming, puede ver la canalización que administra la tabla de streaming en la interfaz de usuario de Delta Live Tables o ver la Información de actualización devuelta por el comando DESCRIBE EXTENDED para la tabla de streaming.

DESCRIBE EXTENDED <table-name>

Ingesta de streaming desde Kafka

Para obtener un ejemplo de ingesta de streaming desde Kafka, consulte read_kafka.

Conceder a los usuarios acceso a una tabla de streaming

Para conceder a los usuarios el privilegio SELECT en la tabla de streaming para que puedan consultarla, pegue lo siguiente en el editor de consultas y, a continuación, haga clic en Ejecutar:

GRANT SELECT ON TABLE <catalog>.<schema>.<table> TO <user-or-group>

Para más información sobre la concesión de privilegios en objetos protegibles de Unity Catalog, consulte Privilegios de Unity Catalog y objetos protegibles.

Recursos adicionales