Registos de erros de qualidade de dados

Quando efetua verificações de qualidade de dados no Catálogo unificado do Microsoft Purview, pode configurar e publicar registos de erros de qualidade de dados para exceções de regras. Esta configuração ajuda-o a rever e corrigir estes registos diretamente no seu armazenamento gerido, seja no Microsoft Fabric Lakehouse ou no Microsoft Azure Data Lake Storage Gen2.

Esta funcionalidade permite que Qualidade de Dados do Microsoft Purview utilizadores identifiquem, analisem e resolvam problemas de qualidade dos dados. Ajuda a garantir a precisão, a exatidão, a consistência, a linha cronológica e a validade dos dados em todos os sistemas e casos de utilização.

Arquitetura

Tem de configurar o armazenamento gerido no Microsoft Purview para publicar registos de erros de qualidade de dados para exceções de regras. O Administrador de Governação de Dados da sua organização é responsável por configurar os detalhes de armazenamento na página de administração do Microsoft Purview. O Responsável pela Qualidade dos Dados pode ativar o botão de alternar a partir da consola Gestão da Qualidade dos Dados ou da página de descrição geral da qualidade dos dados no Catálogo unificado> Aqualidade dos Dados de gestão> de dados.

Localização do armazenamento de configuração

• No Catálogo unificado, aceda a Gestão> de estado de funcionamentoQualidade dos dados.
• Selecione um domínio de governação na lista para abrir a respetiva página de detalhes.
• Selecione Gerir e, em seguida, selecione Definições.
• Selecione Azure Região na lista pendente de todas as regiões de Azure suportadas.
• Depois de selecionar a região Azure, selecione Novo para configurar os detalhes de armazenamento.
• Selecione Tipo de Armazenamento: Recursos de Infraestrutura ou Data Lake Storage Gen2.
• Introduza o URL de Localização.
• Se selecionar Recursos de Infraestrutura como o tipo de armazenamento, introduza o ID da Área de Trabalho e o ID do Lakehouse. Conceda contribuidor acesso ao Microsoft Purview Manage Service Identity (MSI) à sua área de trabalho recursos de infraestrutura.
• Se selecionar Azure Data Lake Storage Gen2 como o tipo de armazenamento, introduza os seguintes detalhes:
  • Adicione Azure subscrição. Para localizar o Azure nome da subscrição, aceda a portal.azure.com.
  • Adicionar Recurso. Para localizar o nome do recurso, selecione Base> selecione Recurso na lista Recursos.
  • Adicione o Nome do contentor . Para localizar o nome do contentor, selecione Data Lake Storage na página de descrição geral emportal.azure.com
  • Adicionar Caminho da Pasta. Estas informações são opcionais.
  • Conceda acesso de Contribuidor de Dados do Blob de Armazenamento à Identidade de Serviço (MSI) de Gestão do Microsoft Purview para o seu contentor de Data Lake Storage Gen2.
• Teste a ligação.
• Guarde a configuração.
• Pode Guardar como rascunho se não tiver todas as informações necessárias para que não perca o que introduziu.

A computação é um pré-requisito para a configuração do ponto final privado ou do armazenamento de rede virtual. Para pedido de aprovação de ponto final privado, selecione a caixa de verificação Ativar V-Net gerido . É enviado um pedido de aprovação para a sua portal do Azure, que tem de aprovar para ativar o acesso à rede virtual. Este processo é semelhante à configuração de uma ligação de Qualidade de Dados. A computação tem de ser implementada na mesma região Azure que a conta de armazenamento.

Se o inquilino dos Recursos de Infraestrutura estiver configurado para ser executado na rede virtual ou numa ligação privada, tem de selecionar a caixa de verificação Ativar V-NET gerido na página Configurar Armazenamento . Em seguida, tem de adicionar o Link Privado ID do Recurso, conforme listado aqui:

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

Para configurar a computação para redes virtuais, veja Configurar a qualidade dos dados para redes virtuais geridas. Para configurar e utilizar a ligação privada para o inquilino do Fabric, siga as instruções para configurar a ligação privada para o inquilino do Fabric antes de configurar a ligação de rede virtual de qualidade de dados e a alocação de computação.

Erro de qualidade de dados regista a definição de armazenamento para armazenamento de rede não virtual – Recursos de infraestrutura:

Erro de qualidade de dados regista a definição de armazenamento para armazenamento virtual – Recursos de infraestrutura:

Erro de qualidade de dados regista a definição de armazenamento para armazenamento não virtual – AdlsG2:

Erro de qualidade de dados regista a definição de armazenamento do armazenamento de rede virtual – AdlsG2:

Deteção de problemas de qualidade de dados

Selecione o recurso de dados e configure as regras para as colunas críticas que precisam de uma medição da Qualidade dos Dados. Pode utilizar regras completas, criar regras personalizadas ou utilizar a funcionalidade de sugestão de regras para recomendar regras para uma análise da qualidade dos dados. Para obter mais informações, veja regras de qualidade de dados.

Ativar a gravação de erros

Para obter registos de erros, tem de ativar a captura e armazenamento de registos de erros para cada tarefa de qualidade de dados executada ao seguir estes passos:

1. Siga as instruções para iniciar uma análise de qualidade de dados num recurso de dados.
2. Depois de selecionar Executar análise de qualidade na página do recurso de dados, ative o botão de alternar para Ativar a publicação de linhas com falhas.

Também pode ativar o registo de erros para todos os recursos de dados com regras de qualidade de dados associadas a produtos de dados ao nível do domínio de governação ao seguir estes passos:

• No Catálogo unificado, aceda a Gestão> de estado de funcionamentoQualidade dos dados.
• Selecione um domínio de governação na lista para abrir a respetiva página de detalhes.
• Selecione Gerir e, em seguida, selecione Definições.
• Ative o botão de alternar para Ativar a publicação de linhas falhadas.

Configurar a agenda para execução

Configure o agendamento para executar a tarefa de qualidade dos dados e publicar os registos de erros no armazenamento configurado. Para obter mais informações, veja Configurar e executar análises de qualidade de dados.

Embora possa executar análises de qualidade de dados numa base ad hoc ao selecionar Executar análise de qualidade numa página de recursos de dados, em cenários de produção é provável que os dados de origem sejam constantemente atualizados. É melhor garantir que a qualidade dos dados é monitorizada regularmente para detetar quaisquer problemas. Automatizar o processo de análise ajuda-o a gerir atualizações regulares de análises de qualidade.

Verificar os registos de erros no armazenamento configurado

1. Localize a localização das linhas de erro publicadas na página de descrição geral da qualidade dos dados.
2. Selecione Ver Relatório de análise, no canto superior direito, imediatamente abaixo da Classificação de qualidade mais recente.
3. Aceda ao caminho apresentado na pasta Data Lake Storage Gen2 ou Fabric Lakehouse, conforme mostrado no exemplo seguinte:

hierarquia de pastas Data Lake Storage Gen2

Data Lake Storage Gen2 contentor (por exemplo, o nome da pasta é DEH):

1. DataQualityAuditExtract
2. Domínio de governação (BusinessDomain)
3. DataProduct
4. DataAsset
5. RunDate
6. RunId
7. Purview DQRunId
8. Ficheiro de registo de erros

Ler registos de erros de qualidade de dados de Data Lake Storage Gen2

Pode utilizar o bloco de notas do Synapse ou do Databricks para ler dados de Data Lake Storage Gen2 contentor com este script do 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 dados de exemplo:

Ler registos de erros do Fabric Lakehouse

1. Aceda ao caminho apresentado no Lakehouse e procure todos os registos de falha publicados no Fabric Lakehouse.
2. Crie um atalho do Delta Parquet para a Tabela Delta. Em alternativa, pode utilizar um ficheiro Delta Parquet para criar uma dashboard dos seus registos de erros de qualidade de dados, como se pode ver neste exemplo:

Pode utilizar um bloco de notas para criar uma tabela de atalhos de registos de recursos de dados com falhas. Veja o script do bloco de notas.

Modelo de dados de saída

Pode ver registos com falhas ao nível de um recurso de dados regido. Para cada nova execução de avaliação com êxito no recurso de dados, o processo adiciona um novo conjunto de registos falhados para cada avaliação de regra à tabela de registos com falhas.

Formato: Delta

Esquema de saída: cada linha de saída contém todas as colunas de uma linha de recurso de dados de entrada que falharam na avaliação de uma regra, juntamente com um conjunto de colunas de metadados que pode utilizar para identificar e analisar a linha com falha.

Chave primária: chave composta das colunas de recursos de dados de entrada + _purviewDQRunId + _purviewDQRuleId

Criar um dashboard do Power BI para os seus registos de erros de qualidade de dados

Utilize o ID do Domínio de Negócio para ligar produtos de dados e os respetivos recursos de dados associados para comunicar registos de erros de qualidade de dados.

• Um domínio empresarial pode ligar a vários produtos de dados.
• Cada produto de dados pode ligar a vários recursos de dados.
• Cada recurso de dados pode ter múltiplas regras.
• Cada regra pode gerar vários registos de erros.

Esta imagem ilustra um modelo de dados para criar um relatório básico do seu registo de erros de qualidade de dados:

Esta imagem ilustra um relatório de exemplo criado com o modelo de dados mostrado na imagem anterior:

Limitações

• São publicados 100 000 registos falhados por regra para cada execução.
• Foram referenciados conjuntos de dados até 100 milhões de linhas com até 40 regras.
• Data Lake Storage Gen2 e o armazenamento de recursos de infraestrutura na rede virtual não são suportados.
• À medida que os dados são armazenados no armazenamento gerido das organizações, a organização detém o controlo de acesso baseado em funções para os dados. A Microsoft publica linhas de erro de qualidade de dados no armazenamento de uma organização com base no consentimento da organização.

Script para criar atalho

Pode automatizar a criação de atalhos de tabela a partir da auditoria de qualidade de dados extrai dados de linha com falhas ao utilizar este script num bloco de notas do Fabric. Depois de criar as tabelas de atalho, os registos de erros da nova tarefa são executados na tabela de atalhos de recursos e são atualizados automaticamente nos Recursos de Infraestrutura.

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

Conteúdo relacionado

• Qualidade de dados para dados do Fabric Lakehouse
• Qualidade de dados para fontes de dados espelhadas do Fabric
• Qualidade de dados para fontes de dados de atalho do Fabric
• Qualidade dos dados para Azure Synapse armazéns de dados e sem servidor
• Qualidade de dados para Catálogo Azure Databricks Unity
• Qualidade dos dados para origens de dados do Snowflake
• Qualidade dos dados do Google BigQuery
• Suporte nativo de qualidade de dados para dados iceberg