Qué hace realmente el destinatario con una tabla compartida
por Jordan Soldo y Nikhil Tallam
Muchas arquitecturas de intercambio de datos comienzan con un patrón conocido:
La parte 1 de este blog describió por qué una tabla compartida puede ser una mejor alternativa a un archivo recurrente o a una fuente de API. Ahora veamos qué hace realmente el destinatario con esa tabla.
Esta guía práctica se centra en el intercambio de Databricks a Databricks (D2D). El destinatario tiene un workspace de Databricks con Unity Catalog habilitado y accede a la tabla compartida a través de un catálogo compartido. El proveedor comparte la tabla con su historial, lo que permite al destinatario utilizarla como origen de Structured Streaming. Volveremos al intercambio de Databricks a Open (D2O) más adelante en la publicación.
Una tabla Delta mantiene un historial ordenado de cada cambio realizado en ella. OpenSharing expone esa tabla a través de un protocolo abierto sin copiarla. Los datos permanecen en el almacenamiento del proveedor y el destinatario los lee en tiempo real.
Por parte del destinatario, se apunta un lector de Structured Streaming a la tabla compartida. Spark registra hasta dónde ha leído en un punto de control (checkpoint), de modo que cada ejecución toma solo los cambios realizados desde la última. El seguimiento de "qué hay de nuevo desde la última vez que consulté" pasa a ser tarea del motor, que se ejecuta sobre datos gobernados manteniendo intactos el linaje y el control de acceso.
Tomemos un ejemplo: un minorista mantiene una tabla de pedidos en tiempo real y la comparte con un destinatario externo que ejecuta Databricks. En lugar de recibir un archivo CSV nocturno, el destinatario puede consultar la tabla de pedidos compartida directamente. Si el destinatario necesita una copia local para análisis o cargas de trabajo posteriores, puede materializar incrementalmente los datos en una tabla Delta. La configuración se realiza una sola vez y se divide entre el proveedor y el destinatario.
Cree la tabla de origen con Change Data Feed activado y publíquela en un recurso compartido WITH HISTORY. El historial es lo que convierte a la tabla en una fuente incremental. Sin él, el destinatario solo obtiene una instantánea estática. Habilitar Change Data Feed (delta.enableChangeDataFeed) es opcional para la lectura incremental principal que se indica a continuación. Actívelo solo si planea usar la ruta de upserts y deletes más adelante. Si usa Databricks Runtime 19 o superior, puede omitir esta propiedad y depender de Change Data Feed automático, que calcula los cambios a nivel de fila en el momento de la consulta para las tablas con seguimiento de filas habilitado y funciona con OpenSharing de Databricks a Databricks.
Registre al destinatario mediante su identificador de recursos compartidos y, a continuación, conceda el acceso:
Tenga en cuenta que CREATE SHARE y CREATE RECIPIENT requieren privilegios a nivel de metastore, no solo permisos de administrador del workspace.
El recurso compartido ahora aparece como un proveedor en su metastore. Móntelo como un catálogo y podrá leerlo como cualquier otra tabla.
Con el recurso compartido montado como shared_in, la lectura se gestiona mediante Structured Streaming.
Trigger.AvailableNow procesa todos los datos disponibles y no procesados actualmente utilizando uno o más micro-lotes y luego finaliza. Esto lo hace ideal para trabajos incrementales programados. En lugar de una programación fija, también puede configurar el trabajo con un activador de actualización de tabla en la tabla compartida, de modo que el trabajo se ejecute solo cuando el proveedor envíe nuevos datos. Se recomienda almacenar el punto de control (checkpoint) en un almacenamiento en la nube duradero como UC Volumes.
Nota: Este ejemplo lee una tabla compartida a través de Delta Sharing, por lo que utiliza ignoreChanges. Si transmite datos directamente desde una tabla Delta normal, las versiones más recientes de Databricks Runtime utilizan skipChangeCommits para el procesamiento de solo adición.
Para comprobar que es realmente incremental, ejecute el flujo una vez con un conjunto inicial de filas. Vuelva a ejecutarlo después de tener nuevas filas en el proveedor. El historial de confirmaciones de la tabla de destino mostrará dos escrituras de streaming:
Version | Operation | Operation Metrics |
|---|---|---|
2 | STREAMING UPDATE | numRemovedFiles: 0, numOutputRows: 200, numOutputBytes: 2307, numAddedFiles: 1 |
1 | STREAMING UPDATE | numRemovedFiles: 0, numOutputRows: 1000, numOutputBytes: 4958, numAddedFiles: 1 |
0 | CREATE TABLE | {} |
La segunda ejecución escribió 200 filas, no 1200. El punto de control (checkpoint) realizó un seguimiento de la última versión leída, por lo que la ejecución extrajo solo las nuevas.
ignoreChanges es adecuado para este patrón de llegada incremental, pero puede volver a emitir filas y no propaga las eliminaciones. Si los consumidores posteriores necesitan inserciones, actualizaciones y eliminaciones, lea en su lugar el Change Data Feed de la tabla compartida:
Cuando se agrega un nuevo cliente, el proveedor no necesita crear un nuevo endpoint SFTP ni implementar un nuevo trabajo de exportación. El proveedor le otorga al destinatario acceso al recurso compartido. El destinatario utiliza el mismo patrón general mientras mantiene separadas sus propias tablas de destino y su lógica de negocio posterior. La diferencia operativa es significativa:
Todo lo anterior funciona porque el destinatario ejecuta Spark en Databricks, que es lo que permite el streaming con puntos de control (checkpoints) y CDF. Otros destinatarios (pandas, herramientas de BI, Snowflake) se conectan a través de OpenSharing con un archivo de credenciales en lugar de un ID de recurso compartido de metastore.
Si ejecutan OSS Spark, la lectura es el mismo código spark.readStream a través del conector de delta-sharing. Sin Spark, no hay un flujo continuo, por lo que extraen los cambios por rango de versiones:
Un primer proyecto útil no necesita resolver todos los escenarios de intercambio. Comience con una tabla Delta normal y un destinatario.
Antes de implementar el flujo, responda a estas preguntas:
Una vez que esas respuestas estén claras, valide el flujo de extremo a extremo más pequeño posible: cree el recurso compartido, lea una instantánea inicial, escriba una nueva fila, actualice una fila, elimine una fila y confirme que el destinatario ve los resultados esperados.
Ese es el cambio práctico que OpenSharing hace posible. En lugar de crear un nuevo flujo de trabajo de transporte e ingesta para cada cliente, puede establecer un contrato de intercambio de datos gobernado y permitir que cada destinatario consuma los datos a través de la ruta que se adapte a su plataforma.
¿Eres nuevo en OpenSharing? Empieza con ¿Qué es OpenSharing? y Crear recursos compartidos para OpenSharing, y luego configura el flujo integral mínimo anterior para una tabla y un destinatario.
(Esta entrada del blog ha sido traducida utilizando herramientas basadas en inteligencia artificial) Publicación original
Suscríbete a nuestro blog y recibe las últimas publicaciones directamente en tu bandeja de entrada.