Das Kopieren von Daten nach und aus Azure Databricks Delta Lake mithilfe von Azure Data Factory oder Azure Synapse Analytics

GILT FÜR: Azure Data Factory Azure Synapse Analytics

In diesem Artikel wird beschrieben, wie Sie die Kopieraktivität in Azure Data Factory und Azure Synapse verwenden, um Daten nach und aus Azure Databricks Delta Lake zu kopieren. Er baut auf dem Artikel zur Kopieraktivität auf, der eine allgemeine Übersicht über die Kopieraktivität enthält.

Unterstützte Funktionen

Dieser Azure Databricks Delta Lake-Connector wird für die folgenden Funktionen unterstützt:

① Azure Integration Runtime ② Selbstgehostete Integration Runtime

Im Allgemeinen bietet der Dienst Unterstützung für Delta Lake mit den folgenden Funktionen, um Ihren verschiedenen Anforderungen zu entsprechen.

• Die Kopieraktivität unterstützt den Azure Databricks Delta Lake-Connector beim Kopieren von Daten aus einem beliebigen unterstützten Quelldatenspeicher in die Azure Databricks Delta Lake-Tabelle und aus der Delta Lake-Tabelle in einen beliebigen unterstützten Senkendatenspeicher. Sie nutzt Ihren Databricks-Cluster zur Durchführung der Datenverschiebung. Weitere Informationen finden Sie im Abschnitt Voraussetzungen.
• Der Zuordnungsdatenfluss unterstützt das generische Delta-Format für Azure Storage als Quelle und Senke zum Lesen und Schreiben von Delta-Dateien für ETL-Vorgänge ohne Code und wird unter der verwalteten Azure Integration Runtime ausgeführt.
• Databricks-Aktivitäten unterstützen zusätzlich zu Delta Lake die Orchestrierung Ihrer codezentrischen ETL- oder Machine Learning-Workload.

Voraussetzungen

Um diesen Azure Databricks Delta Lake-Connector zu verwenden, müssen Sie einen Cluster in Azure Databricks einrichten.

• Zum Kopieren von Daten nach Delta Lake ruft die Kopieraktivität den Azure Databricks-Cluster auf, um Daten aus einem Azure Storage zu lesen, bei dem es sich entweder um Ihre ursprüngliche Quelle oder einen Stagingbereich handelt, in den der Dienst die Quelldaten zunächst über eine integrierte gestaffelte Kopie schreibt. Weitere Informationen finden Sie unter Delta Lake als Senke.
• Ähnlich verhält es sich mit dem Kopieren von Daten aus Delta Lake, wobei die Kopieraktivität den Azure Databricks-Cluster aufruft, um Daten in einen Azure Storage zu schreiben, bei dem es sich entweder um Ihre ursprüngliche Senke oder einen Stagingbereich handelt, von dem aus der Dienst weiterhin Daten über eine integrierte gestaffelte Kopie in die endgültige Senke schreibt. Weitere Informationen finden Sie unter Delta Lake als Quelle.

Der Databricks-Cluster muss Zugriff auf das Azure Blob- oder Azure Data Lake Storage Gen2-Konto haben, sowohl auf den Speichercontainer/das Dateisystem, der/das für Quelle/Senke/Staging verwendet wird, als auch auf den Container/das Dateisystem, in den/das Sie die Delta Lake-Tabellen schreiben möchten.

• Zur Verwendung von Azure Data Lake Storage Gen2 können Sie im Rahmen der Apache Spark-Konfiguration einen Dienstprinzipal auf dem Databricks-Cluster konfigurieren. Führen Sie die Schritte unter Direkter Zugriff mit Dienstprinzipal aus.

• Zur Verwendung von Azure-Blobspeicher können Sie einen Zugriffsschlüssel für das Speicherkonto oder ein SAS-Token auf dem Databricks-Cluster im Rahmen der Apache Spark-Konfiguration konfigurieren. Befolgen Sie die Schritte in Zugreifen auf Azure-Blobspeicher mithilfe der RDD-API.

Wenn der von Ihnen konfigurierte Cluster während der Ausführung der Kopieraktivität beendet wurde, startet der Dienst ihn automatisch. Wenn Sie eine Pipeline mithilfe der Erstellungsbenutzeroberfläche erstellen, benötigen Sie für Vorgänge wie die Datenvorschau einen aktiven Cluster. Der Dienst startet den Cluster nicht in Ihrem Auftrag.

Angeben der Clusterkonfiguration

1. Wählen Sie in der Dropdownliste Clustermodus die Option Standard aus.

2. Wählen Sie in der Dropdownliste Databricks-Runtimeversion eine Databricks-Runtimeversion aus.

3. Aktivieren Sie Automatisches Optimieren, indem Sie die folgenden Eigenschaften zu Ihrer Spark-Konfiguration hinzufügen:

   spark.databricks.delta.optimizeWrite.enabled true
   spark.databricks.delta.autoCompact.enabled true
   

4. Konfigurieren Sie Ihren Cluster je nach Ihren Integrations- und Skalierungsanforderungen.

Weitere Informationen zur Clusterkonfiguration finden Sie unter Konfigurieren von Clustern.

Erste Schritte

Sie können eines der folgenden Tools oder SDKs verwenden, um die Kopieraktivität mit einer Pipeline zu verwenden:

• Das Tool „Daten kopieren“
• Azure-Portal
• Das .NET SDK
• Das Python SDK
• Azure PowerShell
• Die REST-API
• Die Azure Resource Manager-Vorlage

Erstellen eines verknüpften Diensts für Azure Databricks Delta Lake über die Benutzeroberfläche

Verwenden Sie die folgenden Schritte, um einen verknüpften Dienst mit Azure Databricks Delta Lake auf der Azure-Portal-Benutzeroberfläche zu erstellen.

1. Navigieren Sie in Ihrem Azure Data Factory- oder Synapse-Arbeitsbereich zur Registerkarte „Verwalten“, wählen Sie „Verknüpfte Dienste“ aus, und klicken Sie dann auf „Neu“:

   • Azure Data Factory
   • Azure Synapse

   

2. Suchen Sie nach Delta, und wählen Sie den Azure Databricks Delta Lake-Connector aus.

   

3. Konfigurieren Sie die Dienstdetails, testen Sie die Verbindung, und erstellen Sie den neuen verknüpften Dienst.

   

Details zur Connectorkonfiguration

Die folgenden Abschnitte enthalten Details zu Eigenschaften, die zum Definieren von Entitäten speziell für einen Azure Databricks Delta Lake-Connector verwendet werden.

Eigenschaften des verknüpften Diensts

Dieser Azure Databricks Delta Lake-Connector unterstützt die folgenden Authentifizierungstypen. Weitere Informationen finden Sie in den entsprechenden Abschnitten.

• Zugriffstoken
• Authentifizierung mit einer systemseitig zugewiesenen verwalteten Identität
• Authentifizierung mit einer benutzerseitig zugewiesenen verwalteten Identität

Zugriffstoken

Die folgenden Eigenschaften werden für den mit Azure Databricks Delta Lake verknüpften Dienst unterstützt:

Beispiel:

{
    "name": "AzureDatabricksDeltaLakeLinkedService",
    "properties": {
        "type": "AzureDatabricksDeltaLake",
        "typeProperties": {
            "domain": "https://adb-xxxxxxxxx.xx.azuredatabricks.net",
            "clusterId": "<cluster id>",
            "accessToken": {
                "type": "SecureString", 
                "value": "<access token>"
          	}
        }
    }
}

Authentifizierung mit einer systemseitig zugewiesenen verwalteten Identität

Weitere Informationen zu systemseitig zugewiesenen verwalteten Identitäten für Azure-Ressourcen finden Sie unter Systemseitig zugewiesene verwaltete Identität für Azure-Ressourcen.

Führen Sie die folgenden Schritte zum Gewähren von Berechtigungen aus, um die Authentifizierung mit einer systemseitig zugewiesenen verwalteten Identität zu verwenden:

1. Rufen Sie die Informationen zur verwalteten Identität ab, indem Sie den Wert der Objekt-ID der verwalteten Identität kopieren, der zusammen mit Ihrer Data Factory oder Ihrem Synapse-Arbeitsbereich generiert wurde.

2. Erteilen Sie der verwalteten Identität die geeigneten Berechtigungen in Azure Databricks. Im Allgemeinen müssen Sie Ihrer systemseitig zugewiesenen verwalteten Identität in der Zugriffssteuerung (IAM) von Azure Databricks mindestens die Rolle Mitwirkender gewähren.

Die folgenden Eigenschaften werden für den mit Azure Databricks Delta Lake verknüpften Dienst unterstützt:

Beispiel:

{
    "name": "AzureDatabricksDeltaLakeLinkedService",
    "properties": {
        "type": "AzureDatabricksDeltaLake",
        "typeProperties": {
            "domain": "https://adb-xxxxxxxxx.xx.azuredatabricks.net",
            "clusterId": "<cluster id>",
            "workspaceResourceId": "<workspace resource id>"
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

Authentifizierung mit einer benutzerseitig zugewiesenen verwalteten Identität

Weitere Informationen zu benutzerseitig zugewiesenen verwalteten Identitäten für Azure-Ressourcen finden Sie unter benutzerseitig zugewiesene verwaltete Identitäten.

Führen Sie die folgenden Schritte aus, um die Authentifizierung mit einer benutzerseitig zugewiesenen verwalteten Identität zu verwenden:

1. Erstellen Sie eine oder mehrere benutzerseitig zugewiesene verwaltete Identitäten, und gewähren Sie ihnen Berechtigungen in Ihrem Azure Databricks. Im Allgemeinen müssen Sie Ihrer benutzerseitig zugewiesenen verwalteten Identität in der Zugriffssteuerung (IAM) von Azure Databricks mindestens die Rolle Mitwirkender gewähren.

2. Weisen Sie Ihrer Data Factory oder Ihrem Synapse-Arbeitsbereich eine oder mehrere benutzerseitig zugewiesene verwaltete Identitäten zu, und erstellen Sie Anmeldeinformationen für jede benutzerseitig zugewiesene verwaltete Identität.

Die folgenden Eigenschaften werden für den mit Azure Databricks Delta Lake verknüpften Dienst unterstützt:

Beispiel:

{
    "name": "AzureDatabricksDeltaLakeLinkedService",
    "properties": {
        "type": "AzureDatabricksDeltaLake",
        "typeProperties": {
            "domain": "https://adb-xxxxxxxxx.xx.azuredatabricks.net",
            "clusterId": "<cluster id>",
            "credential": {
                "referenceName": "credential1",
                "type": "CredentialReference"
            },
            "workspaceResourceId": "<workspace resource id>"
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

Dataset-Eigenschaften

Eine vollständige Liste mit den Abschnitten und Eigenschaften, die zum Definieren von Datasets zur Verfügung stehen, finden Sie im Artikel zu Datasets.

Folgende Eigenschaften werden für das Azure Databricks Delta Lake-Dataset unterstützt.

Beispiel:

{
    "name": "AzureDatabricksDeltaLakeDataset",
    "properties": {
        "type": "AzureDatabricksDeltaLakeDataset",
        "typeProperties": {
            "database": "<database name>",
            "table": "<delta table name>"
        },
        "schema": [ < physical schema, optional, retrievable during authoring > ],
        "linkedServiceName": {
            "referenceName": "<name of linked service>",
            "type": "LinkedServiceReference"
        }
    }
}

Eigenschaften der Kopieraktivität

Eine vollständige Liste mit den Abschnitten und Eigenschaften zum Definieren von Aktivitäten finden Sie im Artikel Pipelines. Dieser Abschnitt enthält eine Liste der Eigenschaften, die von der Azure Databricks Delta Lake-Quelle und -Senke unterstützt werden.

Delta Lake als Quelle

Beim Kopieren von Daten aus Azure Databricks Delta Lake werden die folgenden Eigenschaften im Abschnitt source der Kopieraktivität unterstützt.

Direkte Kopie aus Delta Lake

Wenn der Senkendatenspeicher und das Format die in diesem Abschnitt beschriebenen Kriterien erfüllen, können Sie mit der Kopieraktivität Kopiervorgänge direkt aus der Azure Databricks Delta-Tabelle in die Senke durchführen. Der Dienst überprüft die Einstellungen und gibt bei der Copy-Aktivitätsausführung einen Fehler aus, wenn die folgenden Kriterien nicht erfüllt werden:

• Der verknüpfte Senkendienst ist Azure-Blobspeicher oder Azure Data Lake Storage Gen2. Die Anmeldeinformation für das Konto sollten in der Clusterkonfiguration von Azure Databricks vorkonfiguriert sein. Weitere Informationen finden Sie unter Voraussetzungen.

• Das Senkendatenformat lautet Parquet, Durch Trennzeichen getrennter Text oder Avro mit den folgenden Konfigurationen und verweist auf einen Ordner anstatt auf eine Datei.

  • Beim Format Parquet wird einer der Komprimierungscodecs None, Snappy oder GZIP verwendet.
  • Beim Format Durch Trennzeichen getrennter Text:
    • rowDelimiter ist ein beliebiges einzelnes Zeichen.
    • compression kann None, BZIP2, GZIP sein.
    • encodingName UTF-7 wird nicht unterstützt.
  • Beim Format Avro wird einer der Komprimierungscodecs None, Deflate oder Snappy verwendet.

• In der Quelle der Kopieraktivität ist additionalColumns nicht angegeben.

• Wenn Daten in durch Trennzeichen getrennter Text kopiert werden, muss fileExtension in der Kopieraktivitätssenke „.csv“ sein.

• In der Zuordnung der Kopieraktivität ist die Typkonvertierung nicht aktiviert.

Beispiel:

"activities":[
    {
        "name": "CopyFromDeltaLake",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<Delta lake input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "AzureDatabricksDeltaLakeSource",
                "sqlReaderQuery": "SELECT * FROM events TIMESTAMP AS OF timestamp_expression"
            },
            "sink": {
                "type": "<sink type>"
            }
        }
    }
]

Gestaffelte Kopie aus Delta Lake

Wenn Ihr Senkendatenspeicher oder das Format nicht den Kriterien für die direkte Kopie entspricht, wie im letzten Abschnitt erwähnt, aktivieren Sie die integrierte gestaffelte Kopie mithilfe einer Azure Storage-Zwischeninstanz. Das gestaffelte Kopieren bietet auch einen höheren Durchsatz. Der Dienst exportiert Daten aus Azure Databricks Delta Lake in den Stagingspeicher, kopiert dann die Daten in die Senke und bereinigt schließlich die temporären Daten aus dem Stagingspeicher. Ausführliche Informationen zum Kopieren von Daten mithilfe von Staging finden Sie unter Gestaffeltes Kopieren.

Um dieses Feature verwenden zu können, erstellen Sie einen mit Azure Blob Storage verknüpften Dienst oder einen mit Azure Data Lake Storage Gen2 verknüpften Dienst, der auf das Speicherkonto als Zwischenbereich verweist. Geben Sie dann die Eigenschaften enableStaging und stagingSettings in der Kopieraktivität an.

Beispiel:

"activities":[
    {
        "name": "CopyFromDeltaLake",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<Delta lake input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "AzureDatabricksDeltaLakeSource",
                "sqlReaderQuery": "SELECT * FROM events TIMESTAMP AS OF timestamp_expression"
            },
            "sink": {
                "type": "<sink type>"
            },
            "enableStaging": true,
            "stagingSettings": {
                "linkedServiceName": {
                    "referenceName": "MyStagingStorage",
                    "type": "LinkedServiceReference"
                },
                "path": "mystagingpath"
            }
        }
    }
]

Delta Lake als Senke

Beim Kopieren von Daten in Azure Databricks Delta Lake werden die folgenden Eigenschaften im Abschnitt sink der Kopieraktivität unterstützt.

Direktes Kopieren nach Delta Lake

Wenn der Quelldatenspeicher und das Format die in diesem Abschnitt beschriebenen Kriterien erfüllen, können Sie mit der Kopieraktivität Kopiervorgänge direkt aus der Quelle in Azure Databricks Delta Lake durchführen. Der Dienst überprüft die Einstellungen und gibt bei der Copy-Aktivitätsausführung einen Fehler aus, wenn die folgenden Kriterien nicht erfüllt werden:

• Der verknüpfte Quelldienst ist Azure-Blobspeicher oder Azure Data Lake Storage Gen2. Die Anmeldeinformation für das Konto sollten in der Clusterkonfiguration von Azure Databricks vorkonfiguriert sein. Weitere Informationen finden Sie unter Voraussetzungen.

• Das Quelldatenformat lautet Parquet, Durch Trennzeichen getrennter Text oder Avro mit den folgenden Konfigurationen und verweist auf einen Ordner anstatt auf eine Datei.

  • Beim Format Parquet wird einer der Komprimierungscodecs None, Snappy oder GZIP verwendet.
  • Beim Format Durch Trennzeichen getrennter Text:
    • rowDelimiter ist „default“ oder ein beliebiges einzelnes Zeichen.
    • compression kann None, BZIP2, GZIP sein.
    • encodingName UTF-7 wird nicht unterstützt.
  • Beim Format Avro wird einer der Komprimierungscodecs None, Deflate oder Snappy verwendet.

• In der Quelle der Kopieraktivität:

  • wildcardFileName enthält nur das Platzhalterzeichen *, aber nicht ?, und wildcardFolderName ist nicht angegeben.
  • prefix, modifiedDateTimeStart, modifiedDateTimeEnd und enablePartitionDiscovery sind nicht angegeben.
  • additionalColumns ist nicht angegeben.

• In der Zuordnung der Kopieraktivität ist die Typkonvertierung nicht aktiviert.

Beispiel:

"activities":[
    {
        "name": "CopyToDeltaLake",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<Delta lake output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "<source type>"
            },
            "sink": {
                "type": "AzureDatabricksDeltaLakeSink",
                "sqlReaderQuery": "VACUUM eventsTable DRY RUN"
            }
        }
    }
]

Gestaffeltes Kopieren nach Delta Lake

Wenn Ihr Quelldatenspeicher oder das Format nicht den Kriterien für die direkte Kopie entspricht, wie im letzten Abschnitt erwähnt, aktivieren Sie die integrierte gestaffelte Kopie mithilfe einer Azure Storage-Zwischeninstanz. Das gestaffelte Kopieren bietet auch einen höheren Durchsatz. Der Dienst konvertiert die Daten automatisch in einen Stagingspeicher, um die Anforderungen an das Datenformat zu erfüllen, und lädt die Daten von dort aus in Delta Lake. Abschließend werden Sie die temporären Daten im Speicher bereinigt. Ausführliche Informationen zum Kopieren von Daten mithilfe von Staging finden Sie unter Gestaffeltes Kopieren.

Um dieses Feature verwenden zu können, erstellen Sie einen mit Azure Blob Storage verknüpften Dienst oder einen mit Azure Data Lake Storage Gen2 verknüpften Dienst, der auf das Speicherkonto als Zwischenbereich verweist. Geben Sie dann die Eigenschaften enableStaging und stagingSettings in der Kopieraktivität an.

Beispiel:

"activities":[
    {
        "name": "CopyToDeltaLake",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<Delta lake output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "<source type>"
            },
            "sink": {
                "type": "AzureDatabricksDeltaLakeSink"
            },
            "enableStaging": true,
            "stagingSettings": {
                "linkedServiceName": {
                    "referenceName": "MyStagingBlob",
                    "type": "LinkedServiceReference"
                },
                "path": "mystagingpath"
            }
        }
    }
]

Überwachung

Es wird der gleiche Ablauf bei der Überwachung von Kopieraktivitäten wie mit anderen Connectors geboten. Da das Laden von Daten von/nach Delta Lake auf Ihrem Azure Databricks-Cluster ausgeführt wird, können Sie darüber hinaus detaillierte Clusterprotokolle anzeigen und die Leistung überwachen.

Eigenschaften der Lookup-Aktivität

Weitere Informationen zu den Eigenschaften finden Sie unter Lookup-Aktivität in Azure Data Factory.

Die Lookup-Aktivität kann bis zu 1000 Zeilen zurückgeben. Wenn das Resultset mehr Datensätze enthält, werden die ersten 1000 Zeilen zurückgegeben.

Zugehöriger Inhalt

Eine Liste der Datenspeicher, die als Quellen und Senken für die Copy-Aktivität unterstützt werden, finden Sie unter Unterstützte Datenspeicher und Formate.