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.

Importante

Las tablas de streaming creadas en Databricks SQL están respaldadas por una canalización de Delta Live Tables sin servidor. El área de trabajo debe admitir canalizaciones sin servidor para usar esta funcionalidad.

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

Establecimiento del canal en tiempo de ejecución

Las tablas de streaming creadas mediante almacenes de SQL se actualizan automáticamente mediante una canalización de Delta Live Tables. Las canalizaciones de Delta Live Tables usan el entorno de ejecución en el current canal de forma predeterminada. Consulte las notas de la versión de Delta Live Tables y el proceso de actualización de la versión para obtener información sobre el proceso de versión.

Databricks recomienda usar el current canal para cargas de trabajo de producción. Las nuevas características se publican por primera vez en el preview canal. Puede establecer una canalización en el canal delta Live Tables de vista previa para probar las nuevas características especificando preview como una propiedad de tabla. Puede especificar esta propiedad al crear la tabla o después de crear la tabla mediante una instrucción ALTER.

En el ejemplo de código siguiente se muestra cómo establecer el canal en versión preliminar en una instrucción CREATE:

CREATE OR REPLACE MATERIALIZED VIEW foo.default.bar
TBLPROPERTIES ('pipelines.channel' = 'preview') as
SELECT
  *
FROM
  range(5)

## <a id="refresh"></a> Refresh a <st> using a DLT pipeline

This section describes patterns for refreshing a <st> with the latest data available from the sources defined in the query.

When you `CREATE` or `REFRESH` a <st>, the update processes using a serverless <DLT> pipeline. Each <st> you define has an associated <DLT> pipeline.

After you run the `REFRESH` command, the DLT pipeline link is returned. You can use the DLT pipeline link to check the status of the refresh.

.. note:: Only the table owner can refresh a <st> to get the latest data. The user that creates the table is the owner, and the owner can't be changed. You might need to refresh your <st> before using [time travel](/delta/history.md#time-travel) queries.

See [_](/delta-live-tables/index.md).

### Ingest new data only

By default, the `read_files` function reads all existing data in the source directory during table creation, and then processes newly arriving records with each refresh.

To avoid ingesting data that already exists in the source directory at the time of table creation, set the `includeExistingFiles` option to `false`. This means that only data that arrives in the directory after table creation is processed. For example:

.. azure::

  ```sql
  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.

Supervisión de ejecuciones mediante el historial de consultas

Puede usar la página del historial de consultas para acceder a los detalles de la consulta y a los perfiles de consulta que pueden ayudarle a identificar consultas con un rendimiento deficiente y cuellos de botella en la canalización delta Live Tables que se usa para ejecutar las actualizaciones de la tabla de streaming. Para obtener información general sobre el tipo de información disponible en los historiales de consultas y los perfiles de consulta, consulte Historial de consultas y Perfil de consulta.

Importante

Esta característica está en versión preliminar pública. Los administradores de áreas de trabajo pueden habilitar esta característica desde la página Versiones preliminares. Consulte Administración de las versiones preliminares de Azure Databricks.

Todas las instrucciones relacionadas con las tablas de streaming aparecen en el historial de consultas. Puede usar el filtro desplegable Instrucción para seleccionar cualquier comando e inspeccionar las consultas relacionadas. Todas las CREATE instrucciones van seguidas de una REFRESH instrucción que se ejecuta de forma asincrónica en una canalización de Delta Live Tables. Las REFRESH instrucciones suelen incluir planes de consulta detallados que proporcionan información sobre la optimización del rendimiento.

Para acceder a REFRESH las instrucciones de la interfaz de usuario del historial de consultas, siga estos pasos:

  1. Haga clic Icono Historial en la barra lateral izquierda para abrir la interfaz de usuario del historial de consultas.
  2. Active la casilla ACTUALIZAR en el filtro desplegable Instrucción .
  3. Haga clic en el nombre de la instrucción de consulta para ver los detalles de resumen, como la duración de la consulta y las métricas agregadas.
  4. Haga clic en Ver perfil de consulta para abrir el perfil de consulta. Consulte Perfil de consulta para obtener más información sobre cómo navegar por el perfil de consulta.
  5. Opcionalmente, puede usar los vínculos de la sección Origen de consulta para abrir la consulta o canalización relacionada.

También puede acceder a los detalles de la consulta mediante vínculos en el editor de SQL o desde un cuaderno asociado a un almacén de SQL.

Nota:

La tabla de streaming debe configurarse para ejecutarse mediante el canal de vista previa . Consulte Establecimiento del canal en tiempo de ejecución.

Recursos adicionales