Kopieren und Transformieren von Daten in Snowflake V1 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 Daten mithilfe der Copy-Aktivität in Azure Data Factory- und Azure Synapse-Pipelines aus und in Snowflake kopieren sowie Daten mithilfe des Tools „Datenfluss“ in Snowflake transformieren. Weitere Informationen finden Sie im Einführungsartikel zu Data Factory oder Azure Synapse Analytics.

Unterstützte Funktionen

Für den Snowflake-Connector werden die folgenden Funktionen unterstützt:

① Azure Integration Runtime ② Selbstgehostete Integration Runtime

Für die Kopieraktivität unterstützt dieser Snowflake-Connector die folgenden Funktionen:

• Kopieren von Daten aus Snowflake unter Verwendung des Snowflake-Befehls COPY into [Speicherort], um optimale Leistung zu erzielen.
• Kopieren von Daten in Snowflake unter Verwendung des Snowflake-Befehls COPY into [Tabelle], um optimale Leistung zu erzielen. Snowflake in Azure wird unterstützt.
• Wenn für einen Proxy eine Verbindung mit Snowflake über eine selbstgehostete Integration Runtime erforderlich ist, müssen Sie die Umgebungsvariablen HTTP_PROXY und HTTPS_PROXY auf dem Integration Runtime-Host konfigurieren.

Voraussetzungen

Wenn sich Ihr Datenspeicher in einem lokalen Netzwerk, in einem virtuellen Azure-Netzwerk oder in einer virtuellen privaten Amazon-Cloud befindet, müssen Sie eine selbstgehostete Integration Runtime konfigurieren, um eine Verbindung herzustellen. Die von der selbstgehosteten Integration Runtime verwendeten IP-Adressen müssen der Positivliste hinzugefügt werden.

Handelt es sich bei Ihrem Datenspeicher um einen verwalteten Clouddatendienst, können Sie die Azure Integration Runtime verwenden. Ist der Zugriff auf IP-Adressen beschränkt, die in den Firewallregeln genehmigt sind, können Sie Azure Integration Runtime-IP-Adressen zur Positivliste hinzufügen.

Das für „Quelle“ oder „Senke“ verwendete Snowflake-Konto benötigt USAGE-Zugriff auf die Datenbank und Lese-/Schreibzugriff auf das Schema und die zugehörigen Tabellen/Sichten. Außerdem muss es die Berechtigung CREATE STAGE für das Schema haben, um die Phase „Extern“ mit SAS-URI erstellen zu können.

Die folgenden Werte für Kontoeigenschaften müssen festgelegt werden

Weitere Informationen zu den von Data Factory unterstützten Netzwerksicherheitsmechanismen und -optionen finden Sie unter Datenzugriffsstrategien.

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 mit Snowflake über die Benutzeroberfläche

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

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

   • Azure Data Factory
   • Azure Synapse

   

2. Suchen Sie nach Snowflake, und wählen Sie dann den Snowflake-Connector aus.

   

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

   

Details zur Connector-Konfiguration

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

Eigenschaften des verknüpften Diensts

Dieser Snowflake-Connector unterstützt die folgenden Authentifizierungstypen. Weitere Informationen finden Sie in den entsprechenden Abschnitten.

• Standardauthentifizierung

Standardauthentifizierung

Die folgenden Eigenschaften werden für den mit Snowflake verknüpften Dienst unterstützt, wenn die Standardauthentifizierung verwendet wird.

Beispiel:

{
    "name": "SnowflakeLinkedService",
    "properties": {
        "type": "Snowflake",
        "typeProperties": {
            "authenticationType": "Basic",
            "connectionString": "jdbc:snowflake://<accountname>.snowflakecomputing.com/?user=<username>&password=<password>&db=<database>&warehouse=<warehouse>&role=<myRole>"
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

Kennwort in Azure Key Vault:

{
    "name": "SnowflakeLinkedService",
    "properties": {
        "type": "Snowflake",
        "typeProperties": {
            "authenticationType": "Basic",
            "connectionString": "jdbc:snowflake://<accountname>.snowflakecomputing.com/?user=<username>&db=<database>&warehouse=<warehouse>&role=<myRole>",
            "password": {
                "type": "AzureKeyVaultSecret",
                "store": { 
                    "referenceName": "<Azure Key Vault linked service name>",
                    "type": "LinkedServiceReference"
                }, 
                "secretName": "<secretName>"
            }
        },
        "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 Snowflake-Dataset unterstützt.

Beispiel:

{
    "name": "SnowflakeDataset",
    "properties": {
        "type": "SnowflakeTable",
        "typeProperties": {
            "schema": "<Schema name for your Snowflake database>",
            "table": "<Table name for your Snowflake database>"
        },
        "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 Snowflake-Quelle und -Senke unterstützt werden.

Snowflake als Quelle

Der Snowflake-Connector verwendet den Snowflake-Befehl COPY into [Speicherort], um optimale Leistung zu erzielen.

Wenn der Senkendatenspeicher und das Format vom Snowflake-Befehl „COPY“ nativ unterstützt werden, können Sie mit der Kopieraktivität Kopiervorgänge direkt aus Snowflake in die Senke durchführen. Details finden Sie unter Direktes Kopieren aus Snowflake. Verwenden Sie andernfalls das integrierte gestaffelte Kopieren aus Snowflake.

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

Direktes Kopieren aus Snowflake

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 Snowflake 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 Blob Storage mit SAS-Authentifizierung. Wenn Sie Daten direkt im folgenden unterstützten Format in Azure Data Lake Storage Gen2 kopieren möchten, können Sie einen verknüpften Azure Blob-Dienst mit SAS-Authentifizierung für Ihr ADLS Gen2-Konto erstellen. Damit vermeiden Sie gestaffelte Kopiervorgänge aus Snowflake.

• Das Senkendatenformat lautet Parquet, Durch Trennzeichen getrennter Text oder JSON mit den folgenden Konfigurationen:

  • Beim Format Parquet wird einer der Komprimierungscodecs None, Snappy oder Lzo verwendet.
  • Beim Format Durch Trennzeichen getrennter Text:
    • rowDelimiter ist \r\n oder ein beliebiges einzelnes Zeichen.
    • compression kann auf keine Komprimierung oder auf gzip, bzip2 oder deflate festgelegt sein.
    • encodingName wird als Standardwert übernommen oder ist auf utf-8 festgelegt.
    • quoteChar ist als doppeltes Anführungszeichen, einfaches Anführungszeichen oder als leere Zeichenfolge (kein Anführungszeichen) festgelegt.
  • Für das JSON-Format unterstützt direktes Kopieren nur den Fall, dass die Snowflake-Quelltabelle oder das Abfrageergebnis nur eine einzelne Spalte hat und der Datentyp dieser Spalte VARIANT, OBJECT oder ARRAY ist.
    • compression kann auf keine Komprimierung oder auf gzip, bzip2 oder deflate festgelegt sein.
    • encodingName wird als Standardwert übernommen oder ist auf utf-8 festgelegt.
    • filePattern in der Senke der Kopieraktivität wird auf dem Standardwert belassen oder auf setOfObjects festgelegt.

• In der Quelle der Copy-Aktivität ist additionalColumns nicht angegeben.

• Die Spaltenzuordnung ist nicht angegeben.

Beispiel:

"activities":[
    {
        "name": "CopyFromSnowflake",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<Snowflake input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "SnowflakeSource",
                "query": "SELECT * FROM MYTABLE",
                "exportSettings": {
                    "type": "SnowflakeExportCopyCommand",
                    "additionalCopyOptions": {
                        "MAX_FILE_SIZE": "64000000",
                        "OVERWRITE": true
                    },
                    "additionalFormatOptions": {
                        "DATE_FORMAT": "'MM/DD/YYYY'"
                    }
                }
            },
            "sink": {
                "type": "<sink type>"
            }
        }
    }
]

Gestaffeltes Kopieren aus Snowflake

Wenn der Senkendatenspeicher oder das Format nicht nativ mit dem Snowflake-Befehl „COPY“ kompatibel ist (wie im letzten Abschnitt beschrieben), aktivieren Sie mithilfe einer Azure Blob Storage-Zwischeninstanz das integrierte gestaffelte Kopieren. Das Feature für gestaffeltes Kopieren bietet Ihnen auch einen höheren Durchsatz. Der Dienst exportiert Daten aus Snowflake 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 diese Funktion verwenden zu können, erstellen Sie einen mit Azure Blob Storage verknüpften Dienst, der auf das Azure Storage-Konto als Stagingzwischenspeicher verweist. Geben Sie dann die Eigenschaften enableStaging und stagingSettings in der Kopieraktivität an.

Beispiel:

"activities":[
    {
        "name": "CopyFromSnowflake",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<Snowflake input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "SnowflakeSource",               
                "query": "SELECT * FROM MyTable",
                "exportSettings": {
                    "type": "SnowflakeExportCopyCommand"
                }
            },
            "sink": {
                "type": "<sink type>"
            },
            "enableStaging": true,
            "stagingSettings": {
                "linkedServiceName": {
                    "referenceName": "MyStagingBlob",
                    "type": "LinkedServiceReference"
                },
                "path": "mystagingpath"
            }
        }
    }
]

Snowflake als Senke

Der Snowflake-Connector verwendet den Snowflake-Befehl COPY into [Tabelle], um optimale Leistung zu erzielen. Er unterstützt das Schreiben von Daten in Snowflake in Azure.

Wenn der Quelldatenspeicher und das Format vom Snowflake-Befehl „COPY“ nativ unterstützt werden, können Sie mit der Kopieraktivität Kopiervorgänge direkt aus der Quelle in Snowflake durchführen. Details finden Sie unter Direktes Kopieren in Snowflake. Verwenden Sie andernfalls das integrierte gestaffelte Kopieren in Snowflake.

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

Direktes Kopieren in Snowflake

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 Snowflake 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 Blob Storage mit SAS-Authentifizierung. Wenn Sie Daten direkt im folgenden unterstützten Format aus Azure Data Lake Storage Gen2 kopieren möchten, können Sie einen verknüpften Azure Blob-Dienst mit SAS-Authentifizierung für Ihr ADLS Gen2-Konto erstellen. Damit vermeiden Sie gestaffelte Kopiervorgänge nach Snowflake.

• Das Quelldatenformat lautet Parquet, Durch Trennzeichen getrennter Text oder JSON mit den folgenden Konfigurationen:

  • Beim Format Parquet wird einer der Komprimierungscodecs None oder Snappy verwendet.

  • Beim Format Durch Trennzeichen getrennter Text:

    • rowDelimiter ist \r\n oder ein beliebiges einzelnes Zeichen. Wenn das Zeilentrennzeichen nicht „\r\n“ ist, muss firstRowAsHeader auf false festgelegt sein. skipLineCount ist nicht angegeben.
    • compression kann auf keine Komprimierung oder auf gzip, bzip2 oder deflate festgelegt sein.
    • encodingName wird als Standardwert übernommen oder ist auf „UTF-8“, „UTF-16“, „UTF-16BE“, „UTF-32“, „UTF-32BE“, „BIG5“, „EUC-JP“, „EUC-KR“, „GB18030“, „ISO-2022-JP“, „ISO-2022-KR“, „ISO-8859-1“, „ISO-8859-2“, „ISO-8859-5“, „ISO-8859-6“, „ISO-8859-7“, „ISO-8859-8“, „ISO-8859-9“, „WINDOWS-1250“, „WINDOWS-1251“, „WINDOWS-1252“, „WINDOWS-1253“, „WINDOWS-1254“ oder „WINDOWS-1255“ festgelegt.
    • quoteChar ist als doppeltes Anführungszeichen, einfaches Anführungszeichen oder als leere Zeichenfolge (kein Anführungszeichen) festgelegt.

  • Für das JSON-Format unterstützt direktes Kopieren nur den Fall, dass die Snowflake-Senkentabelle nur eine einzelne Spalte hat und der Datentyp dieser Spalte VARIANT, OBJECT oder ARRAY ist.

    • compression kann auf keine Komprimierung oder auf gzip, bzip2 oder deflate festgelegt sein.
    • encodingName wird als Standardwert übernommen oder ist auf utf-8 festgelegt.
    • Die Spaltenzuordnung ist nicht angegeben.

• In der Quelle der Kopieraktivität:

  • additionalColumns ist nicht angegeben.
  • Wenn es sich bei der Quelle um einen Ordner handelt, wird recursive auf TRUE festgelegt.
  • prefix, modifiedDateTimeStart, modifiedDateTimeEnd und enablePartitionDiscovery wurden nicht angegeben.

Beispiel:

"activities":[
    {
        "name": "CopyToSnowflake",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<Snowflake output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "<source type>"
            },
            "sink": {
                "type": "SnowflakeSink",
                "importSettings": {
                    "type": "SnowflakeImportCopyCommand",
                    "copyOptions": {
                        "FORCE": "TRUE",
                        "ON_ERROR": "SKIP_FILE"
                    },
                    "fileFormatOptions": {
                        "DATE_FORMAT": "YYYY-MM-DD"
                    }
                }
            }
        }
    }
]

Gestaffeltes Kopieren in Snowflake

Wenn der Quelldatenspeicher oder das Format nicht nativ mit dem Snowflake-Befehl „COPY“ kompatibel ist (wie im letzten Abschnitt beschrieben), aktivieren Sie mithilfe einer Azure Blob Storage-Zwischeninstanz das integrierte gestaffelte Kopieren. Das Feature für gestaffeltes Kopieren bietet Ihnen auch einen höheren Durchsatz. Der Dienst konvertiert die Daten automatisch, damit das Datenformat den Anforderungen von Snowflake entspricht. Dann wird der Befehl „COPY“ aufgerufen, um die Daten in Snowflake zu laden. Abschließend werden Sie die temporären Daten in Blob Storage bereinigt. Ausführliche Informationen zum Kopieren von Daten mithilfe von Staging finden Sie unter Gestaffeltes Kopieren.

Um diese Funktion verwenden zu können, erstellen Sie einen mit Azure Blob Storage verknüpften Dienst, der auf das Azure Storage-Konto als Stagingzwischenspeicher verweist. Geben Sie dann die Eigenschaften enableStaging und stagingSettings in der Kopieraktivität an.

Beispiel:

"activities":[
    {
        "name": "CopyToSnowflake",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<Snowflake output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "<source type>"
            },
            "sink": {
                "type": "SnowflakeSink",
                "importSettings": {
                    "type": "SnowflakeImportCopyCommand"
                }
            },
            "enableStaging": true,
            "stagingSettings": {
                "linkedServiceName": {
                    "referenceName": "MyStagingBlob",
                    "type": "LinkedServiceReference"
                },
                "path": "mystagingpath"
            }
        }
    }
]

Eigenschaften von Mapping Data Flow

Beim Transformieren von Daten im Zuordnungsdatenfluss können Sie in Snowflake Tabellen lesen und in diese schreiben. Weitere Informationen finden Sie unter Quellentransformation und Senkentransformation in Zuordnungsdatenflüssen. Sie können ein Snowflake-Dataset oder Inlinedataset als Quell- und Senkentyp verwenden.

Quellentransformation

In der folgenden Tabelle sind die von einer Snowflake-Quelle unterstützten Eigenschaften aufgeführt. Sie können diese Eigenschaften auf der Registerkarte Quelloptionen bearbeiten. Der Connector verwendet die interne Datenübertragung von Snowflake.

Beispiele für Snowflake-Quellskripts

Wenn Sie das Snowflake-Dataset als Quelltyp verwenden, sieht das zugehörige Datenflussskript wie folgt aus:

source(allowSchemaDrift: true,
	validateSchema: false,
	query: 'select * from MYTABLE',
	format: 'query') ~> SnowflakeSource

Wenn Sie ein Inlinedataset verwenden, sieht das zugehörige Datenflussskript wie folgt aus:

source(allowSchemaDrift: true,
	validateSchema: false,
	format: 'query',
	query: 'select * from MYTABLE',
	store: 'snowflake') ~> SnowflakeSource

Native Änderungsnachverfolgung

Azure Data Factory unterstützt jetzt ein natives Feature in Snowflake, das als Änderungsnachverfolgung bekannt ist, was das Nachverfolgen von Änderungen in Form von Protokollen umfasst. Dieses Feature von Snowflake ermöglicht es uns, die Änderungen der Daten im Zeitverlauf zu verfolgen, was für das inkrementelle Laden von Daten und für Überprüfungszwecke nützlich ist. Wenn Sie dieses Feature verwenden möchten, erstellen wir beim Aktivieren von Change Data Capture und Auswählen der Snowflake-Änderungsnachverfolgung ein Stream-Objekt für die Quelltabelle, das die Änderungsnachverfolgung in der Snowflake-Tabelle ermöglicht. Anschließend verwenden wir die CHANGES-Klausel in unserer Abfrage, um nur die neuen oder aktualisierten Daten aus der Quelltabelle abzurufen. Außerdem wird empfohlen, die Pipeline so zu planen, dass Änderungen innerhalb des Intervalls der Datenaufbewahrungszeit für die Snowflake-Quelltabelle verarbeitet werden, sonst kann der Benutzer inkonsistentes Verhalten bei erfassten Änderungen sehen.

Senkentransformation

In der folgenden Tabelle sind die von einer Snowflake-Senke unterstützten Eigenschaften aufgeführt. Sie können diese Eigenschaften auf der Registerkarte Einstellungen bearbeiten. Bei Verwendung eines Inlinedatasets werden zusätzliche Einstellungen angezeigt. Diese entsprechen den Eigenschaften, die im Abschnitt zu den Dataseteigenschaften beschrieben sind. Der Connector verwendet die interne Datenübertragung von Snowflake.

Beispiele für Snowflake-Senkenskripts

Wenn Sie das Snowflake-Dataset als Senkentyp verwenden, sieht das zugehörige Datenflussskript wie folgt aus:

IncomingStream sink(allowSchemaDrift: true,
	validateSchema: false,
	deletable:true,
	insertable:true,
	updateable:true,
	upsertable:false,
	keys:['movieId'],
	format: 'table',
	skipDuplicateMapInputs: true,
	skipDuplicateMapOutputs: true) ~> SnowflakeSink

Wenn Sie ein Inlinedataset verwenden, sieht das zugehörige Datenflussskript wie folgt aus:

IncomingStream sink(allowSchemaDrift: true,
	validateSchema: false,
	format: 'table',
	tableName: 'table',
	schemaName: 'schema',
	deletable: true,
	insertable: true,
	updateable: true,
	upsertable: false,
	store: 'snowflake',
	skipDuplicateMapInputs: true,
	skipDuplicateMapOutputs: true) ~> SnowflakeSink

Abfrage-Pushdown-Optimierung

Durch Festlegen des Protokolliergrads der Pipeline auf „Keine“ schließen wir die Übertragung von Zwischenmetriken für die Transformation aus, wodurch potenzielle Hindernisse für Spark-Optimierungen vermieden und die von Snowflake bereitgestellte Abfrage-Pushdown-Optimierung ermöglicht werden. Diese Pushdown-Optimierung ermöglicht erhebliche Leistungsverbesserungen für große Snowflake-Tabellen mit umfangreichen Datasets.

Eigenschaften der Lookup-Aktivität

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

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.