Data Quality-Fehlerdatensätze

Wenn Sie Datenqualitätsprüfungen in Microsoft Purview Unified Catalog durchführen, können Sie Datenqualitätsfehlerdatensätze für Regelausnahmen konfigurieren und veröffentlichen. Dieses Setup hilft Ihnen, diese Datensätze direkt in Ihrem verwalteten Speicher zu überprüfen und zu beheben, unabhängig davon, ob in Microsoft Fabric Lakehouse oder Microsoft Azure Data Lake Storage Gen2.

Dieses Feature ermöglicht es Microsoft Purview Data Quality Benutzern, Probleme mit der Datenqualität zu identifizieren, zu analysieren und zu beheben. Dadurch wird die Datengenauigkeit, Vollständigkeit, Konsistenz, Aktualität und Gültigkeit über Systeme und Anwendungsfälle hinweg sichergestellt.

Architektur

Sie müssen Ihren verwalteten Speicher in Microsoft Purview einrichten, um Datenqualitätsfehlerdatensätze für Regelausnahmen zu veröffentlichen. Der Data Governance-Administrator für Ihre organization ist für die Konfiguration der Speicherdetails auf der Microsoft Purview-Verwaltungsseite verantwortlich. Der Data Quality Steward kann die Umschaltfläche entweder über die Data Quality Manage-Konsole oder die Übersichtsseite der Datenqualität unter Unified Catalog>Integritätsverwaltung>Datenqualität aktivieren.

Konfigurationsspeicherort

• Wechseln Sie Unified Catalog zu Integritätsverwaltung>Datenqualität.
• Wählen Sie eine Governancedomäne aus der Liste aus, um die zugehörige Detailseite zu öffnen.
• Wählen Sie Verwalten und dann Einstellungen aus.
• Wählen Sie Azure Region aus der Dropdownliste aller unterstützten Azure Regionen aus.
• Nachdem Sie die Azure Region ausgewählt haben, wählen Sie Neu aus, um Speicherdetails zu konfigurieren.
• Wählen Sie Speichertyp: Fabric oder Data Lake Storage Gen2 aus.
• Geben Sie die Standort-URL ein.
• Wenn Sie Fabric als Speichertyp auswählen, geben Sie die Arbeitsbereichs-ID und die Lakehouse-ID ein. Gewähren Sie Mitwirkender Zugriff auf Microsoft Purview Manage Service Identity (MSI) auf Ihren Fabric-Arbeitsbereich.
• Wenn Sie Azure Data Lake Storage Gen2 als Speichertyp auswählen, geben Sie die folgenden Details ein:
  • Fügen Sie Azure Abonnement hinzu. Navigieren Sie zu , um den Namen Ihres Azure Abonnements zu findenportal.azure.com.
  • Ressource hinzufügen. Um den Ressourcennamen zu ermitteln, wählen SieStart> ressource aus der Liste Ressource aus.
  • Containername hinzufügen. Um den Containernamen zu ermitteln, wählen Sie Data Lake Storage auf der Übersichtsseite inportal.azure.com
  • Ordnerpfad hinzufügen. Dies sind optionale Informationen.
  • Gewähren Sie Storage Blob Data Contributor Zugriff auf Microsoft Purview Manage Service Identity (MSI) auf Ihren Data Lake Storage Gen2 Container.
• Testen Sie die Verbindung.
• Speichern Sie die Konfiguration.
• Sie können Als Entwurf speichern , wenn Sie nicht über alle erforderlichen Informationen verfügen, sodass Sie nicht verlieren, was Sie eingegeben haben.

Compute ist eine Voraussetzung für die Einrichtung von privatem Endpunkt oder virtuellem Netzwerkspeicher. Aktivieren Sie für die Genehmigungsanforderung für private Endpunkte das Kontrollkästchen Verwaltetes V-Net aktivieren . Eine Genehmigungsanforderung wird an Ihre Azure-Portal gesendet, die Sie genehmigen müssen, um den Zugriff auf virtuelle Netzwerke zu ermöglichen. Dieser Vorgang ähnelt dem Einrichten einer Data Quality-Verbindung. Das Compute muss in derselben Azure Region wie das Speicherkonto bereitgestellt werden.

Wenn Ihr Fabric-Mandant für die Ausführung in einem virtuellen Netzwerk oder in einer privaten Verbindung konfiguriert ist, müssen Sie auf der Seite Speicher konfigurieren das Kontrollkästchen Verwaltetes V-NET aktivieren aktivieren. Anschließend müssen Sie die Private Link Ressourcen-ID wie hier aufgeführt hinzufügen:

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

Informationen zum Einrichten von Compute für virtuelle Netzwerke finden Sie unter Einrichten der Datenqualität für verwaltete virtuelle Netzwerke. Befolgen Sie zum Einrichten und Verwenden von Private Link für Ihren Fabric-Mandanten die Anweisungen zum Einrichten der privaten Verbindung für den Fabric-Mandanten , bevor Sie die Virtuelle Netzwerkverbindung mit Datenqualität und die Computezuordnung konfigurieren.

Data Quality error records storage setting for non-virtual network storage – Fabric:

Data Quality error records storage setting for virtual storage – Fabric:

Data Quality error records storage setting for non-virtual storage – AdlsG2:

Data Quality error records storage setting for virtual network storage – AdlsG2:

Erkennung von Datenqualitätsproblemen

Wählen Sie die Datenressource aus, und konfigurieren Sie Regeln für die kritischen Spalten, die eine Data Quality-Messung erfordern. Sie können sofort einsatzbereite Regeln verwenden, benutzerdefinierte Regeln erstellen oder das Regelvorschlagsfeature verwenden, um Regeln für eine Data Quality-Überprüfung zu empfehlen. Weitere Informationen finden Sie unter Datenqualitätsregeln.

Aktivieren der Fehleraufzeichnung

Um Fehlerdatensätze abzurufen, müssen Sie die Erfassung und Speicherung von Fehlerdatensätzen für jeden ausgeführten Data Quality-Auftrag aktivieren, indem Sie die folgenden Schritte ausführen:

1. Befolgen Sie die Anweisungen, um eine Datenqualitätsüberprüfung für eine Datenressource zu starten.
2. Nachdem Sie auf der Seite der Datenressource Qualitätsüberprüfung ausführen ausgewählt haben, aktivieren Sie den Umschalter für Veröffentlichung fehlerhafter Zeilen aktivieren.

Sie können die Fehleraufzeichnung auch für alle Datenassets mit Datenqualitätsregeln aktivieren, die Datenprodukten in einer Governancedomänenebene zugeordnet sind, indem Sie die folgenden Schritte ausführen:

• Wechseln Sie Unified Catalog zu Integritätsverwaltung>Datenqualität.
• Wählen Sie eine Governancedomäne aus der Liste aus, um die zugehörige Detailseite zu öffnen.
• Wählen Sie Verwalten und dann Einstellungen aus.
• Aktivieren Sie die Umschaltfläche für Die Veröffentlichung fehlerhafter Zeilen aktivieren.

Einrichten des Zeitplans für die Ausführung

Richten Sie den Zeitplan ein, um Ihren Data Quality-Auftrag auszuführen und die Fehlerdatensätze im konfigurierten Speicher zu veröffentlichen. Weitere Informationen finden Sie unter Konfigurieren und Ausführen von Datenqualitätsüberprüfungen.

Obwohl Sie Datenqualitätsüberprüfungen ad-hoc ausführen können, indem Sie auf einer Datenassetseite Qualitätsüberprüfung ausführen auswählen, ist es in Produktionsszenarien wahrscheinlich, dass die Quelldaten ständig aktualisiert werden. Es ist am besten, sicherzustellen, dass die Datenqualität regelmäßig überwacht wird, um Probleme zu erkennen. Durch die Automatisierung des Überprüfungsprozesses können Sie regelmäßige Updates von Qualitätsscans verwalten.

Überprüfen der Fehlerdatensätze im konfigurierten Speicher

1. Suchen Sie den Speicherort der veröffentlichten Fehlerzeilen auf der Übersichtsseite der Datenqualität.
2. Wählen Sie in der oberen rechten Ecke direkt unterhalb der Aktuellen Qualitätsbewertung die Option Scanbericht anzeigen aus.
3. Wechseln Sie zu dem Pfad, der in Ihrem Data Lake Storage Gen2- oder Fabric Lakehouse-Ordner angezeigt wird, wie im folgenden Beispiel gezeigt:

Data Lake Storage Gen2 Ordnerhierarchie

Data Lake Storage Gen2 Container (der Ordnername lautet z. B. DEH):

1. DataQualityAuditExtract
2. Governancedomäne (BusinessDomain)
3. DataProduct
4. DataAsset
5. RunDate
6. RunId
7. Purview DQRunId
8. Fehlerdatensatzdatei

Lesen von Datenqualitätsfehlerdatensätzen aus Data Lake Storage Gen2

Sie können das Synapse- oder Databricks-Notebook verwenden, um Daten aus Data Lake Storage Gen2 Container zu lesen, indem Sie dieses PySpark-Skript verwenden:

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

Beispieldataset:

Lesen von Fehlerdatensätzen aus Fabric Lakehouse

1. Wechseln Sie zu dem pfad, der in Ihrem Lakehouse angezeigt wird, und durchsuchen Sie alle Fehlerdatensätze, die in Fabric Lakehouse veröffentlicht wurden.
2. Erstellen Sie eine Verknüpfung von Delta Parquet zu Delta Table. Alternativ können Sie eine Delta Parquet-Datei verwenden, um eine Dashboard Ihrer Datenqualitätsfehlerdatensätze zu erstellen, wie in diesem Beispiel gezeigt:

Sie können ein Notebook verwenden, um eine Verknüpfungstabelle mit fehlerhaften Datensätzen von Datenassetobjekten zu erstellen. Weitere Informationen finden Sie im Notebookskript.

Ausgabedatenmodell

Sie können fehlerhafte Datensätze auf der Ebene einer verwalteten Datenressource anzeigen. Für jede neue Bewertungsausführung, die für die Datenressource erfolgreich ist, fügt der Prozess einen neuen Satz fehlerhafter Datensätze für jede Regelauswertung zur Tabelle mit fehlerhaften Datensätzen hinzu.

Format: Delta

Ausgabeschema: Jede Ausgabezeile enthält alle Spalten einer Eingabedatenobjektzeile, bei der eine Regelauswertung fehlgeschlagen ist, sowie einen Satz von Metadatenspalten, mit denen Sie die fehlerhafte Zeile identifizieren und analysieren können.

Primärschlüssel: Zusammengesetzter Schlüssel aus Eingabedatenobjektspalten + _purviewDQRunId + _purviewDQRuleId

Erstellen einer Power BI-Dashboard für Ihre Datenqualitätsfehlerdatensätze

Verwenden Sie die Geschäftsdomänen-ID, um Datenprodukte und die zugehörigen Datenressourcen zu verknüpfen, um Datenqualitätsfehlerdatensätze zu melden.

• Eine Geschäftsdomäne kann mit mehreren Datenprodukten verknüpft werden.
• Jedes Datenprodukt kann mit mehreren Datenassets verknüpft werden.
• Jede Datenressource kann über mehrere Regeln verfügen.
• Jede Regel kann mehrere Fehlerdatensätze generieren.

Diese Abbildung veranschaulicht ein Datenmodell zum Erstellen eines einfachen Berichts Ihres Datenqualitätsfehlerdatensatzes:

Diese Abbildung veranschaulicht einen Beispielbericht, der mit dem datenmodell erstellt wurde, das in der vorherigen Abbildung dargestellt ist:

Begrenzungen

• Pro Regel werden für jede Ausführung 100.000 fehlerhafte Datensätze veröffentlicht.
• Datasets mit bis zu 100 Millionen Zeilen mit bis zu 40 Regeln wurden als Benchmarks bewertet.
• Data Lake Storage Gen2 und Fabric-Speicher im virtuellen Netzwerk werden nicht unterstützt.
• Da Daten im verwalteten Speicher von Organisationen gespeichert werden, besitzt die organization die rollenbasierte Zugriffssteuerung für die Daten. Microsoft veröffentlicht Datenqualitätsfehlerzeilen im Speicher eines organization basierend auf der Zustimmung des organization.

Skript zum Erstellen der Verknüpfung

Mithilfe dieses Skripts in einem Fabric-Notebook können Sie die Erstellung von Tabellenverknüpfungen aus der Datenqualitätsüberwachung automatisieren. Nach dem Erstellen der Verknüpfungstabellen werden Ihre Fehlerdatensätze für einen neuen Auftrag in der Verknüpfungstabelle des Medienobjekts automatisch in Fabric aktualisiert.

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

Verwandte Inhalte

• Datenqualität für Fabric Lakehouse-Daten
• Datenqualität für gespiegelte Fabric-Datenquellen
• Datenqualität für Fabric-Verknüpfungsdatenquellen
• Datenqualität für Azure Synapse serverlose und Data Warehouses
• Datenqualität für Azure Databricks Unity Catalog
• Datenqualität für Snowflake-Datenquellen
• Datenqualität für Google BigQuery
• Native Unterstützung der Datenqualität für Iceberg-Daten