Enregistrements d’erreurs de qualité des données

Lorsque vous effectuez des vérifications de la qualité des données dans Catalogue unifié Microsoft Purview, vous pouvez configurer et publier des enregistrements d’erreurs de qualité des données pour les exceptions de règle. Cette configuration vous permet d’examiner et de corriger ces enregistrements directement dans votre stockage managé, que ce soit dans Microsoft Fabric Lakehouse ou Microsoft Azure Data Lake Storage Gen2.

Cette fonctionnalité permet aux utilisateurs Qualité des données Microsoft Purview d’identifier, d’analyser et de résoudre les problèmes de qualité des données. Il permet de garantir l’exactitude, l’exhaustivité, la cohérence, l’actualité et la validité des données sur les systèmes et les cas d’usage.

Architecture

Vous devez configurer votre stockage managé dans Microsoft Purview pour publier des enregistrements d’erreurs de qualité des données pour les exceptions de règle. L’administrateur de gouvernance des données de votre organization est responsable de la configuration des détails du stockage dans la page d’administration de Microsoft Purview. Le gestionnaire de la qualité des données peut activer le bouton bascule à partir de la console Gestion de la qualité des données ou de la page vue d’ensemble de la qualité des données dans Catalogue unifié>Gestion de la santé Qualité des>données.

Emplacement de stockage de la configuration

• Dans Catalogue unifié, accédez à Gestion de> l’intégritéQualité des données.
• Sélectionnez un domaine de gouvernance dans la liste pour ouvrir sa page de détails.
• Sélectionnez Gérer, puis Paramètres.
• Sélectionnez Azure Région dans la liste déroulante de toutes les régions Azure prises en charge.
• Après avoir sélectionné la région Azure, sélectionnez Nouveau pour configurer les détails du stockage.
• Sélectionnez Type de stockage : Infrastructure ou Data Lake Storage Gen2.
• Entrez l’URL de l’emplacement.
• Si vous sélectionnez Fabric comme type de stockage, entrez l’ID d’espace de travail et l’ID Lakehouse. Accordez contributeur accès à Microsoft Purview Manage Service Identity (MSI) à votre espace de travail Fabric.
• Si vous sélectionnez Azure Data Lake Storage Gen2 comme type de stockage, entrez les détails suivants :
  • Ajoutez Azure abonnement. Pour trouver le nom de votre abonnement Azure, accédez à portal.azure.com.
  • Ajouter une ressource. Pour trouver le nom de la ressource, sélectionnez Accueil> , sélectionnez Ressource dans la liste Ressource.
  • Ajouter le nom du conteneur . Pour trouver le nom du conteneur, sélectionnez Data Lake Storage dans la page vue d’ensemble dansportal.azure.com
  • Ajoutez le chemin d’accès au dossier. Il s’agit d’informations facultatives.
  • Accordez à Contributeur de données Blob de stockage l’accès à Microsoft Purview Manage Service Identity (MSI) à votre conteneur Data Lake Storage Gen2.
• Testez la connexion.
• Enregistrez la configuration.
• Vous pouvez Enregistrer en tant que brouillon si vous ne disposez pas de toutes les informations requises afin de ne pas perdre ce que vous avez entré.

Le calcul est requis pour la configuration du point de terminaison privé ou du stockage de réseau virtuel. Pour la demande d’approbation de point de terminaison privé, cochez la case Activer le réseau virtuel managé . Une demande d’approbation est envoyée à votre Portail Azure, que vous devez approuver pour activer l’accès au réseau virtuel. Ce processus est similaire à la configuration d’une connexion de qualité des données. Le calcul doit être déployé dans la même région Azure que le compte de stockage.

Si votre locataire Fabric est configuré pour s’exécuter sur un réseau virtuel ou dans une liaison privée, vous devez cocher la case Activer le réseau virtuel managé dans la page Configurer le stockage . Vous devez ensuite ajouter l’ID de ressource Private Link comme indiqué ici :

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

Pour configurer le calcul pour les réseaux virtuels, consultez Configurer la qualité des données pour les réseaux virtuels managés. Pour configurer et utiliser une liaison privée pour votre locataire Fabric, suivez les instructions pour configurer la liaison privée pour le locataire Fabric avant de configurer la connexion au réseau virtuel de qualité des données et l’allocation de calcul.

Paramètre de stockage des enregistrements d’erreur de qualité des données pour le stockage non-réseau virtuel - Fabric :

Paramètre de stockage des enregistrements d’erreur de qualité des données pour le stockage virtuel - Fabric :

Paramètre de stockage de l’erreur de qualité des données pour le stockage non virtuel - AdlsG2 :

Paramètre de stockage d’erreur de qualité des données pour le stockage de réseau virtuel - AdlsG2 :

Détection des problèmes de qualité des données

Sélectionnez la ressource de données et configurez des règles pour les colonnes critiques qui nécessitent une mesure de la qualité des données. Vous pouvez utiliser des règles prêtes à l’emploi, créer des règles personnalisées ou utiliser la fonctionnalité de suggestion de règle pour recommander des règles pour une analyse de la qualité des données. Pour plus d’informations, consultez Règles de qualité des données.

Activer l’enregistrement des erreurs

Pour obtenir les enregistrements d’erreur, vous devez activer la capture et le stockage des enregistrements d’erreurs pour chaque travail de qualité des données exécuté en procédant comme suit :

1. Suivez les instructions pour démarrer une analyse de la qualité des données sur une ressource de données.
2. Après avoir sélectionné Exécuter l’analyse de qualité sur la page de la ressource de données, activez le bouton bascule pour Activer la publication des lignes ayant échoué.

Vous pouvez également activer l’enregistrement des erreurs pour toutes les ressources de données avec des règles de qualité des données associées aux produits de données dans un niveau de domaine de gouvernance en procédant comme suit :

• Dans Catalogue unifié, accédez à Gestion de> l’intégritéQualité des données.
• Sélectionnez un domaine de gouvernance dans la liste pour ouvrir sa page de détails.
• Sélectionnez Gérer, puis Paramètres.
• Activez le bouton bascule pour Activer la publication des lignes ayant échoué.

Configurer la planification pour l’exécution

Configurez la planification pour exécuter votre travail de qualité des données et publier les enregistrements d’erreurs dans le stockage configuré. Pour plus d’informations, consultez Configurer et exécuter des analyses de qualité des données.

Bien que vous puissiez exécuter des analyses de la qualité des données sur une base ad hoc en sélectionnant Exécuter l’analyse de qualité sur une page de ressource de données, dans les scénarios de production, il est probable que les données sources soient constamment mises à jour. Il est préférable de s’assurer que la qualité des données est régulièrement surveillée afin de détecter les éventuels problèmes. L’automatisation du processus d’analyse vous permet de gérer les mises à jour régulières des analyses qualité.

Vérifier les enregistrements d’erreur dans le stockage configuré

1. Recherchez l’emplacement des lignes d’erreur publiées à partir de la page vue d’ensemble de la qualité des données.
2. Sélectionnez Afficher le rapport d’analyse, en haut à droite, juste en dessous du dernier score de qualité.
3. Accédez au chemin d’accès indiqué dans votre dossier Data Lake Storage Gen2 ou Fabric Lakehouse, comme indiqué dans l’exemple suivant :

hiérarchie de dossiers Data Lake Storage Gen2

Data Lake Storage Gen2 conteneur (par exemple, le nom du dossier est DEH) :

1. DataQualityAuditExtract
2. Domaine de gouvernance (BusinessDomain)
3. DataProduct
4. DataAsset
5. RunDate
6. RunId
7. Purview DQRunId
8. Fichier d’enregistrement d’erreur

Lire les enregistrements d’erreur de qualité des données à partir de Data Lake Storage Gen2

Vous pouvez utiliser synapse ou le notebook Databricks pour lire des données à partir de Data Lake Storage Gen2 conteneur à l’aide de ce script 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) 

Exemple de jeu de données :

Lire les enregistrements d’erreur à partir de Fabric Lakehouse

1. Accédez au chemin d’accès indiqué dans votre Lakehouse et parcourez tous les enregistrements d’échec publiés dans Fabric Lakehouse.
2. Créez un raccourci de Delta Parquet à Delta Table. Vous pouvez également utiliser un fichier Delta Parquet pour créer un tableau de bord de vos enregistrements d’erreurs de qualité des données, comme illustré dans cet exemple :

Vous pouvez utiliser un notebook pour créer une table de raccourcis d’enregistrements de ressources de données ayant échoué. Consultez le script de notebook.

Modèle de données de sortie

Vous pouvez afficher les enregistrements ayant échoué au niveau d’une ressource de données régie. Pour chaque nouvelle exécution d’évaluation qui réussit sur la ressource de données, le processus ajoute un nouvel ensemble d’enregistrements ayant échoué pour chaque évaluation de règle à la table des enregistrements ayant échoué.

Format : Delta

Schéma de sortie : chaque ligne de sortie contient toutes les colonnes d’une ligne de ressource de données d’entrée qui a échoué à une évaluation de règle, ainsi qu’un ensemble de colonnes de métadonnées que vous pouvez utiliser pour identifier et analyser la ligne ayant échoué.

Clé primaire : clé composite des colonnes de ressources de données d’entrée + _purviewDQRunId + _purviewDQRuleId

Créer un tableau de bord Power BI pour vos enregistrements d’erreurs de qualité des données

Utilisez l’ID de domaine métier pour lier les produits de données et leurs ressources de données associées afin de signaler les enregistrements d’erreurs de qualité des données.

• Un domaine d’entreprise peut être lié à plusieurs produits de données.
• Chaque produit de données peut être lié à plusieurs ressources de données.
• Chaque ressource de données peut avoir plusieurs règles.
• Chaque règle peut générer plusieurs enregistrements d’erreur.

Cette image illustre un modèle de données pour créer un rapport de base de votre enregistrement d’erreur de qualité des données :

Cette image illustre un exemple de rapport créé avec le modèle de données indiqué dans l’image précédente :

Limitations

• 100 000 enregistrements ayant échoué sont publiés par règle pour chaque exécution.
• Des jeux de données allant jusqu’à 100 millions de lignes avec jusqu’à 40 règles ont été testés.
• Data Lake Storage Gen2 et le stockage Fabric dans un réseau virtuel ne sont pas pris en charge.
• Comme les données sont stockées dans le stockage managé des organisations, le organization possède le contrôle d’accès en fonction du rôle pour les données. Microsoft publie des lignes d’erreur de qualité des données dans le stockage d’une organization en fonction du consentement de l’organization.

Script pour créer un raccourci

Vous pouvez automatiser la création de raccourcis de table à partir de l’audit de qualité des données extrait les données de ligne ayant échoué à l’aide de ce script dans un notebook Fabric. Après avoir créé les tables de raccourcis, vos enregistrements d’erreur pour les nouvelles exécutions de travaux sur la table de raccourcis de ressources sont automatiquement actualisés dans 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)    

Contenu connexe

• Qualité des données pour les données Fabric Lakehouse
• Qualité des données des sources de données en miroir Fabric
• Qualité des données des sources de données de raccourci Fabric
• Qualité des données pour Azure Synapse entrepôts de données et serverless
• Qualité des données pour le catalogue Unity Databricks d’Azure
• Qualité des données pour les sources de données Snowflake
• Qualité des données pour Google BigQuery
• Prise en charge native de la qualité des données pour les données Iceberg