El problema que estás experimentando se debe a que las columnas de identidad en las tablas dinámicas delta de Databricks no funcionan exactamente como en SQL Server.
En Databricks, las columnas de identidad se generan durante la ingesta de datos, pero no se actualizan automáticamente cuando se agregan nuevos datos a la tabla.
Para solucionar este problema, puedes intentar lo siguiente:
row_number()
en lugar de identity()
para generar un número único para cada fila.CREATE OR REFRESH STREAMING LIVE TABLE my_dlt (
dlt_id BIGINT,
source_column1 STRING,
source_column2 STRING
) TBLPROPERTIES (
delta.enableChangeDataFeed = true,
"quality" = "silver"
)
AS
WITH stream_input AS (
SELECT DISTINCT source_column1, source_column2
FROM stream(source_catalog.bronze_schema.source_table)
)
SELECT
row_number() OVER (ORDER BY source_column1) as dlt_id,
source_column1,
source_column2
FROM stream_input;
dlt_id
sea una columna de identidad que se incrementa automáticamente, puedes utilizar la función monotonically_increasing_id()
en combinación con la función row_number()
.CREATE OR REFRESH STREAMING LIVE TABLE my_dlt (
dlt_id BIGINT,
source_column1 STRING,
source_column2 STRING
) TBLPROPERTIES (
delta.enableChangeDataFeed = true,
"quality" = "silver"
)
AS
WITH stream_input AS (
SELECT DISTINCT source_column1, source_column2
FROM stream(source_catalog.bronze_schema.source_table)
)
SELECT
monotonically_increasing_id() + row_number() OVER (ORDER BY source_column1) as dlt_id,
source_column1,
source_column2
FROM stream_input;
Espero que esto te ayude a resolver el problema. ¡Si tienes alguna otra pregunta, no dudes en preguntar!