Registros de errores de calidad de datos

Al realizar comprobaciones de calidad de datos en Catálogo unificado de Microsoft Purview, puede configurar y publicar registros de errores de calidad de datos para excepciones de regla. Esta configuración le ayuda a revisar y corregir estos registros directamente en el almacenamiento administrado, ya sea en Microsoft Fabric Lakehouse o Microsoft Azure Data Lake Storage Gen2.

Esta característica permite a Calidad de datos de Microsoft Purview usuarios identificar, analizar y abordar problemas de calidad de datos. Ayuda a garantizar la precisión de los datos, la integridad, la coherencia, la escala de tiempo y la validez entre sistemas y casos de uso.

Arquitectura

Debe configurar el almacenamiento administrado en Microsoft Purview para publicar registros de errores de calidad de datos para excepciones de regla. El administrador de gobernanza de datos de su organización es responsable de configurar los detalles del almacenamiento en la página de administración de Microsoft Purview. Data Quality Steward puede habilitar el botón de alternancia desde la consola de Administración de calidad de datos o desde la página de información general sobre la calidad de los datos en Catálogo unificado> La calidad de los datos de administración de>estado.

Ubicación de almacenamiento de configuración

• En Catálogo unificado, vaya a Calidad de los datos de administración de>estado.
• Seleccione un dominio de gobernanza de la lista para abrir su página de detalles.
• Seleccione Administrar y, a continuación, seleccione Configuración.
• Seleccione Azure Región en la lista desplegable de todas las regiones Azure admitidas.
• Después de seleccionar la región Azure, seleccione Nuevo para configurar los detalles del almacenamiento.
• Seleccione Tipo de almacenamiento: Tejido o Data Lake Storage Gen2.
• Escriba la dirección URL de ubicación.
• Si selecciona Fabric como tipo de almacenamiento, escriba el identificador del área de trabajo y el identificador de Lakehouse. Conceda acceso de colaborador a Microsoft Purview Manage Service Identity (MSI) a su área de trabajo de Fabric.
• Si selecciona Azure Data Lake Storage Gen2 como tipo de almacenamiento, escriba los detalles siguientes:
  • Agregar Azure suscripción. Para buscar el nombre de la suscripción Azure, vaya a portal.azure.com.
  • Agregar recurso. Para buscar el nombre del recurso, seleccione Inicio>, seleccione Recurso en la lista Recurso.
  • Agregue el nombre del contenedor . Para buscar el nombre del contenedor, seleccione Data Lake Storage en la página de información general deportal.azure.com
  • Agregar ruta de acceso de carpeta. Se trata de información opcional.
  • Conceda acceso al colaborador de datos de Storage Blob a Microsoft Purview Manage Service Identity (MSI) al contenedor de Data Lake Storage Gen2.
• Pruebe la conexión.
• Guarde la configuración.
• Puede guardar como borrador si no tiene toda la información necesaria para que no pierda lo que ha escrito.

El proceso es un requisito previo para la configuración del punto de conexión privado o del almacenamiento de red virtual. Para la solicitud de aprobación de punto de conexión privado, active la casilla Habilitar V-Net administrado . Se envía una solicitud de aprobación a la Azure Portal, que debe aprobar para habilitar el acceso a la red virtual. Este proceso es similar a la configuración de una conexión de calidad de datos. El proceso debe implementarse en la misma región Azure que la cuenta de almacenamiento.

Si el inquilino de Fabric está configurado para ejecutarse en una red virtual o en un vínculo privado, debe seleccionar la casilla Habilitar V-NET administrada en la página Configurar almacenamiento . A continuación, debe agregar el identificador de recurso de Private Link como se muestra aquí:

/subscriptions/07d669d6-83f2-4f15-8b5b 4a4b31a7050e/resourceGroups/pdgbugbashfabricvnet/providers/Microsoft.PowerBI/priva teLinkServicesForPowerBI/fabricvnetpl.

Para configurar el proceso para redes virtuales, consulte Configuración de la calidad de los datos para redes virtuales administradas. Para configurar y usar private link para el inquilino de Fabric, siga las instrucciones para configurar el vínculo privado para el inquilino de Fabric antes de configurar la conexión de red virtual de calidad de datos y la asignación de proceso.

Configuración de almacenamiento de registros de errores de calidad de datos para almacenamiento de red no virtual: Fabric:

Configuración de almacenamiento de registros de errores de calidad de datos para el almacenamiento virtual: Tejido:

Configuración de almacenamiento de registros de errores de calidad de datos para almacenamiento no virtual: AdlsG2:

Configuración de almacenamiento de registros de errores de calidad de datos para el almacenamiento de red virtual: AdlsG2:

Detección de problemas de calidad de datos

Seleccione el recurso de datos y configure reglas para las columnas críticas que necesitan una medición de calidad de datos. Puede usar reglas integradas, crear reglas personalizadas o usar la característica de sugerencia de reglas para recomendar reglas para un examen de calidad de datos. Para obtener más información, consulte reglas de calidad de datos.

Habilitación de la grabación de errores

Para obtener registros de errores, debe habilitar la captura y el almacenamiento de registros de error para cada trabajo de calidad de datos que se ejecute siguiendo estos pasos:

1. Siga las instrucciones para iniciar un examen de calidad de datos en un recurso de datos.
2. Después de seleccionar Ejecutar examen de calidad en la página del recurso de datos, active el botón de alternancia Habilitar publicación de filas con errores.

También puede habilitar la grabación de errores para todos los recursos de datos con reglas de calidad de datos asociadas a productos de datos en un nivel de dominio de gobernanza siguiendo estos pasos:

• En Catálogo unificado, vaya a Calidad de los datos de administración de>estado.
• Seleccione un dominio de gobernanza de la lista para abrir su página de detalles.
• Seleccione Administrar y, a continuación, seleccione Configuración.
• Active el botón de alternancia habilitar la publicación de filas con errores.

Configuración de la programación para ejecutar

Configure la programación para ejecutar el trabajo de calidad de datos y publicar los registros de error en el almacenamiento configurado. Para obtener más información, consulte configuración y ejecución de exámenes de calidad de datos.

Aunque puede ejecutar exámenes de calidad de datos de forma ad hoc seleccionando Ejecutar examen de calidad en una página de recursos de datos, en escenarios de producción es probable que los datos de origen se actualicen constantemente. Es mejor asegurarse de que la calidad de los datos se supervisa periódicamente para detectar cualquier problema. La automatización del proceso de examen le ayuda a administrar actualizaciones periódicas de los exámenes de calidad.

Comprobación de los registros de error en el almacenamiento configurado

1. Busque la ubicación de las filas de error publicadas en la página de información general sobre la calidad de los datos.
2. Seleccione Ver informe de examen, en la esquina superior derecha justo debajo de la puntuación de calidad Más reciente.
3. Vaya a la ruta de acceso que se muestra en la carpeta Data Lake Storage Gen2 o Fabric Lakehouse, como se muestra en el ejemplo siguiente:

jerarquía de carpetas Data Lake Storage Gen2

Data Lake Storage Gen2 contenedor (por ejemplo, el nombre de la carpeta es DEH):

1. DataQualityAuditExtract
2. Dominio de gobernanza (BusinessDomain)
3. DataProduct
4. DataAsset
5. RunDate
6. RunId
7. Purview DQRunId
8. Archivo de registro de errores

Leer registros de errores de calidad de datos de Data Lake Storage Gen2

Puede usar el cuaderno de Synapse o Databricks para leer datos de Data Lake Storage Gen2 contenedor mediante este script de PySpark:

#For complete error records of all jobs run on the asset
path = "abfss://TestWSPDGBB@onelake.dfs.fabric.microsoft.com/TestSelfServeLakehouse.Lakehouse/Files/SelfServeFolder/DataQualityAuditExtracts/BusinessDomain_577b8e54-e534-478d-bbc6-19a3221fc71e/DataProduct_a04c82a2-2372-4b9e-9e0a-4dfd8959ee85/DataAsset_4b2b2644-c94a-447f-9b2e-32961ac0170b"
df = spark.read.format("delta").load(path)
display(df)

#For All Runs on a particular Date
path = "abfss://TestWSPDGBB@onelake.dfs.fabric.microsoft.com/TestSelfServeLakehouse.Lakehouse/Files/SelfServeFolder/DataQualityAuditExtracts/BusinessDomain_577b8e54-e534-478d-bbc6-19a3221fc71e/DataProduct_a04c82a2-2372-4b9e-9e0a-4dfd8959ee85/DataAsset_4b2b2644-c94a-447f-9b2e-32961ac0170b"
df = spark.read.format("delta").load(path).where("_purviewDQRunDate = '2025-08-06'")
display(df)

#For a Specific run
path = "abfss://TestWSPDGBB@onelake.dfs.fabric.microsoft.com/TestSelfServeLakehouse.Lakehouse/Files/SelfServeFolder/DataQualityAuditExtracts/BusinessDomain_577b8e54-e534-478d-bbc6-19a3221fc71e/DataProduct_a04c82a2-2372-4b9e-9e0a-4dfd8959ee85/DataAsset_4b2b2644-c94a-447f-9b2e-32961ac0170b"
df = spark.read.format("delta").load(path).where("_purviewDQRunId = '647b8a8f-7140-4158-9914-9c2409a5ec67'")
display(df)

#For a specific rule of the run
path = "abfss://TestWSPDGBB@onelake.dfs.fabric.microsoft.com/TestSelfServeLakehouse.Lakehouse/Files/SelfServeFolder/DataQualityAuditExtracts/BusinessDomain_577b8e54-e534-478d-bbc6-19a3221fc71e/DataProduct_a04c82a2-2372-4b9e-9e0a-4dfd8959ee85/DataAsset_4b2b2644-c94a-447f-9b2e-32961ac0170b"
df = spark.read.format("delta").load(path).where("_purviewDQRunId = '647b8a8f-7140-4158-9914-9c2409a5ec67' AND _purviewDQRuleId = '32d57f8d-87dc-4862-90c9-c3bcad1051ad'")
display(df) 

#For a specific rule across all runs
path = "abfss://TestWSPDGBB@onelake.dfs.fabric.microsoft.com/TestSelfServeLakehouse.Lakehouse/Files/SelfServeFolder/DataQualityAuditExtracts/BusinessDomain_577b8e54-e534-478d-bbc6-19a3221fc71e/DataProduct_a04c82a2-2372-4b9e-9e0a-4dfd8959ee85/DataAsset_4b2b2644-c94a-447f-9b2e-32961ac0170b"
df = spark.read.format("delta").load(path).where(" _purviewDQRuleId = '32d57f8d-87dc-4862-90c9-c3bcad1051ad'")
display(df) 

#For comparing two runs
path = "abfss://TestWSPDGBB@onelake.dfs.fabric.microsoft.com/TestSelfServeLakehouse.Lakehouse/Files/SelfServeFolder/DataQualityAuditExtracts/BusinessDomain_577b8e54-e534-478d-bbc6-19a3221fc71e/DataProduct_a04c82a2-2372-4b9e-9e0a-4dfd8959ee85/DataAsset_4b2b2644-c94a-447f-9b2e-32961ac0170b"
df = spark.read.format("delta").load(path).where("_purviewDQRunId = '647b8a8f-7140-4158-9914-9c2409a5ec67' OR _purviewDQRunId = 'a508f960-254c-4eb1-9097-03307402475b'")
display(df) 

Conjunto de datos de ejemplo:

Leer registros de errores de Fabric Lakehouse

1. Vaya a la ruta de acceso que se muestra en Lakehouse y examine todos los registros de error publicados en Fabric Lakehouse.
2. Cree un acceso directo de Delta Parquet a Delta Table. O bien, puede usar un archivo Delta Parquet para crear un panel de los registros de errores de calidad de datos, como se muestra en este ejemplo:

Puede usar un cuaderno para crear una tabla de accesos directos de registros de recursos de datos con errores. Consulte el script del cuaderno.

Modelo de datos de salida

Puede ver los registros con errores en el nivel de un recurso de datos gobernado. Para cada nueva ejecución de evaluación que se realice correctamente en el recurso de datos, el proceso agrega un nuevo conjunto de registros con errores para cada evaluación de reglas a la tabla de registros con errores.

Formato: Delta

Esquema de salida: cada fila de salida contiene todas las columnas de una fila de recursos de datos de entrada que no pudieron evaluar una regla, junto con un conjunto de columnas de metadatos que puede usar para identificar y analizar la fila con errores.

Clave principal: clave compuesta de columnas de recursos de datos de entrada + _purviewDQRunId + _purviewDQRuleId

Creación de un panel de Power BI para los registros de error de calidad de datos

Use el identificador de dominio empresarial para vincular los productos de datos y sus recursos de datos asociados para notificar registros de errores de calidad de datos.

• Un dominio empresarial puede vincularse a varios productos de datos.
• Cada producto de datos puede vincularse a varios recursos de datos.
• Cada recurso de datos puede tener varias reglas.
• Cada regla puede generar varios registros de error.

Esta imagen muestra un modelo de datos para crear un informe básico del registro de errores de calidad de datos:

Esta imagen muestra un informe de ejemplo creado con el modelo de datos que se muestra en la imagen anterior:

Limitaciones

• Se publican 100 000 registros con errores por regla para cada ejecución.
• Se han realizado pruebas comparativas de conjuntos de datos de hasta 100 millones de filas con hasta 40 reglas.
• no se admiten Data Lake Storage Gen2 ni el almacenamiento de Fabric en la red virtual.
• A medida que los datos se almacenan en el almacenamiento administrado de las organizaciones, la organización posee el control de acceso basado en rol para los datos. Microsoft publica filas de error de calidad de datos en el almacenamiento de una organización en función del consentimiento de la organización.

Script para crear acceso directo

Puede automatizar la creación de accesos directos de tabla a partir de la auditoría de calidad de datos que extrae datos de fila con errores mediante este script en un cuaderno de Fabric. Después de crear las tablas de accesos directos, los registros de error del nuevo trabajo se ejecutan en la tabla de accesos directos de recursos se actualizan automáticamente en Fabric.

# NEW Script to automate the creation of table shortcuts from Data Quality audit extracts failed row data.

# Update these three values for your Fabric Workspace ID, Lakehouse ID and Purview BYOC Self-serve store path
workspaceId = "f28dc1c8-360c-4788-9f46-e69853b1c40d" #Example: f28dc1c8-360c-4788-9f46-e69853b1c40d
lakehouseId = "77d6df6b-64ab-4628-985f-9365591a85a8" #Example: 77d6df6b-64ab-4628-985f-9365591a85a8
dataqualityauditPath = "Files/SelfServeFolder/DataQualityAuditExtracts" #Example: "Files/DEH2/DataQualityAuditExtracts"

#Use for lakehouses with Lakehouse-Schema (Public preview feature)
DestinationShortcutUri = f"abfss://{workspaceId}@onelake.dfs.fabric.microsoft.com/{lakehouseId}/Tables/DataQualityAuditExtracts" 

#Use for lakehouses without Lakehouse-Schema
#DestinationShortcutUri = f"abfss://{workspaceId}@onelake.dfs.fabric.microsoft.com/{lakehouseId}/Tables" 

import sempy.fabric as fabric
from sempy.fabric.exceptions import FabricHTTPException
import fnmatch
import re
import os
from collections import deque

SourceUri = f"abfss://{workspaceId}@onelake.dfs.fabric.microsoft.com/{lakehouseId}/{dataqualityauditPath}"

request_headers = {
            "Authorization": "Bearer " + mssparkutils.credentials.getToken("pbi"),
            "Content-Type": "application/json"
        }
print(request_headers)


def is_delta_table(uri: str):
    #print("Checking for uri:" + uri)
    delta_log_path = os.path.join(uri, "_delta_log")
    return mssparkutils.fs.exists(delta_log_path)

def extract_onelake_https_uri_components(uri):
    pattern = re.compile(r"abfss://([^@]+)@[^/]+/([^/]+)/(.*)")
    match = pattern.search(uri)
    if match:
        workspace_id, item_id, path = match.groups()
        return workspace_id, item_id, path
    else:
        return None, None, None

def is_valid_onelake_uri(uri: str) -> bool:
    workspace_id, item_id, path = extract_onelake_https_uri_components(uri)
    if "abfss://" not in uri or workspace_id is None or item_id is None or path is None:
        return False
    return True

def get_onelake_shorcut(workspace_id: str, item_id: str, path: str, name: str):
    shortcut_uri = (
        f"v1/workspaces/{workspace_id}/items/{item_id}/shortcuts/{path}/{name}"
    )
    result = client.get(shortcut_uri).json()
    return result

def get_last_path_segment(uri: str):
    path = uri.split("/")  # Split the entire URI by '/'
    return path[-1] if path else None

def create_onelake_shorcut(SourceUri: str, dest_uri: str, result: list):
    src_workspace_id, src_item_id, src_path = extract_onelake_https_uri_components(
        SourceUri
    )

    dest_workspace_id, dest_item_id, dest_path = extract_onelake_https_uri_components(
        dest_uri
    )

    name = get_last_path_segment(SourceUri)
    dest_uri_joined = os.path.join(dest_uri, name)

    # If the destination path already exists, return without creating shortcut
    if mssparkutils.fs.exists(dest_uri_joined):
        print(f"Table already exists: {dest_uri_joined}")
        result.append(dest_uri_joined)
        return None

    request_body = {
        "name": name,
        "path": dest_path,
        "target": {
            "oneLake": {
                "itemId": src_item_id,
                "path": src_path,
                "workspaceId": src_workspace_id,
            }
        },
    }
    #print(request_body)

    shortcut_uri = f"v1/workspaces/{dest_workspace_id}/items/{dest_item_id}/shortcuts"
    print(f"Creating shortcut: {shortcut_uri}/{name}..")
    try:
        client.post(shortcut_uri, json=request_body, headers=request_headers)
    except FabricHTTPException as e:
        print(e)
        print(e.error_reason)
        return None

    return get_onelake_shorcut(dest_workspace_id, dest_item_id, dest_path, name)


client = fabric.FabricRestClient()

queue = deque([SourceUri])
result = []

while queue:
    current_uri = queue.popleft()

    #print(current_uri)
    
    if  is_delta_table(os.path.join("", current_uri)):
        #print(current_uri)
        shortcut = create_onelake_shorcut(os.path.join("", current_uri), DestinationShortcutUri, result)
        if shortcut is not None:
            result.append(shortcut)
        continue;
    
    # List subitems in the current folder
    subitems = mssparkutils.fs.ls(current_uri)   
    for item in subitems:
        if item.isDir:
            queue.append(item.path)   


print(f"{len(result)}" + " error record shortcut tables present in the lakehouse: ")
for item in result:
    print(item)    

Contenido relacionado

• Calidad de los datos de Fabric Lakehouse
• Calidad de datos para orígenes de datos reflejados en Fabric
• Calidad de datos para orígenes de datos de acceso directo de Fabric
• Calidad de datos para Azure Synapse almacenamiento de datos y sin servidor
• Calidad de los datos para Unity Catalog para Azure Databricks
• Calidad de los datos para orígenes de datos de Snowflake
• Calidad de los datos de Google BigQuery
• Compatibilidad nativa con la calidad de los datos de Iceberg