Adatok másolása és átalakítása a Snowflake V1-ben az Azure Data Factory vagy az Azure Synapse Analytics használatával

A következőkre vonatkozik: Azure Data Factory Azure Synapse Analytics

Ez a cikk bemutatja, hogyan használhatja a Copy tevékenység az Azure Data Factoryben és az Azure Synapse-folyamatokban az adatok Snowflake-ből és Snowflake-be való másolásához, valamint Adatfolyam a Snowflake-adatok átalakításához. További információkért tekintse meg a Data Factory vagy az Azure Synapse Analytics bevezető cikkét.

Támogatott képességek

Ez a Snowflake-összekötő a következő képességeket támogatja:

(1) Azure-integrációs modul (2) Saját üzemeltetésű integrációs modul

A Copy tevékenység ez a Snowflake-összekötő a következő funkciókat támogatja:

• A Snowflake COPY parancsát használó Snowflake-adatok másolása a [location] parancsba a legjobb teljesítmény elérése érdekében.
• Másolja az adatokat a Snowflake-be, amely kihasználja a Snowflake COPY in [table] parancsát a legjobb teljesítmény elérése érdekében. Támogatja a Snowflake-t az Azure-ban.
• Ha proxyra van szükség a Snowflake-hez való csatlakozáshoz egy saját üzemeltetésű integrációs modulból, konfigurálnia kell a környezeti változókat HTTP_PROXY és HTTPS_PROXY az integrációs modul gazdagépén.

Előfeltételek

Ha az adattár helyszíni hálózaton, Azure-beli virtuális hálózaton vagy Amazon Virtual Private Cloudon belül található, konfigurálnia kell egy saját üzemeltetésű integrációs modult a csatlakozáshoz. Mindenképpen adja hozzá a saját üzemeltetésű integrációs modul által használt IP-címeket az engedélyezett listához.

Ha az adattár felügyelt felhőalapú adatszolgáltatás, használhatja az Azure Integration Runtime-ot. Ha a hozzáférés a tűzfalszabályokban jóváhagyott IP-címekre korlátozódik, felveheti az Azure Integration Runtime IP-eket az engedélyezett listára.

A Forráshoz vagy fogadóhoz használt Snowflake-fióknak rendelkeznie kell a szükséges USAGE hozzáféréssel az adatbázishoz, és olvasási/írási hozzáféréssel kell rendelkeznie a sémán és az alatta lévő táblákon/nézeteken. Emellett a sémának is rendelkeznie CREATE STAGE kell a külső szakasz SAS URI-val való létrehozásához.

A következő fióktulajdonság-értékeket kell megadni

A Data Factory által támogatott hálózati biztonsági mechanizmusokkal és lehetőségekkel kapcsolatos további információkért lásd az adathozzáférési stratégiákat.

Első lépések

A Copy tevékenység folyamattal való végrehajtásához használja az alábbi eszközök vagy SDK-k egyikét:

• Az Adatok másolása eszköz
• Az Azure Portal
• A .NET SDK
• A Python SDK
• Azure PowerShell
• A REST API
• Az Azure Resource Manager-sablon

Társított szolgáltatás létrehozása a Snowflake-hez felhasználói felülettel

Az alábbi lépésekkel létrehozhat egy társított szolgáltatást a Snowflake-hez az Azure Portal felhasználói felületén.

1. Keresse meg az Azure Data Factory vagy a Synapse-munkaterület Kezelés lapját, és válassza a Társított szolgáltatások lehetőséget, majd kattintson az Új gombra:

   • Azure Data Factory
   • Azure Synapse

   

2. Keresse meg a Snowflake-t, és válassza a Snowflake-összekötőt.

   

3. Konfigurálja a szolgáltatás részleteit, tesztelje a kapcsolatot, és hozza létre az új társított szolgáltatást.

   

Az összekötő konfigurációjának részletei

A következő szakaszok a Snowflake-összekötőhöz tartozó entitásokat definiáló tulajdonságok részleteit ismertetik.

Társított szolgáltatás tulajdonságai

Ez a Snowflake-összekötő a következő hitelesítési típusokat támogatja. Részletekért tekintse meg a megfelelő szakaszokat.

• Alapszintű hitelesítés

Alapszintű hitelesítés

A Snowflake társított szolgáltatás esetében az alábbi tulajdonságok támogatottak az alapszintű hitelesítés használatakor.

Példa:

{
    "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"
        }
    }
}

Jelszó az Azure Key Vaultban:

{
    "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"
        }
    }
}

Adathalmaz tulajdonságai

Az adathalmazok meghatározásához elérhető szakaszok és tulajdonságok teljes listáját az Adathalmazok című cikkben találja.

A Snowflake-adatkészlet esetében az alábbi tulajdonságok támogatottak.

Példa:

{
    "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"
        }
    }
}

Másolási tevékenység tulajdonságai

A tevékenységek meghatározásához elérhető szakaszok és tulajdonságok teljes listáját a Folyamatok című cikkben találja. Ez a szakasz a Snowflake-forrás és -fogadó által támogatott tulajdonságok listáját tartalmazza.

Hópehely mint forrás

A Snowflake-összekötő a Snowflake COPY into [location] parancsát használja a legjobb teljesítmény eléréséhez.

Ha a Snowflake COPY parancs natív módon támogatja a fogadó adattárát és formátumát, a Copy tevékenység használatával közvetlenül másolhatja a Snowflake-ből a fogadóba. További részletekért lásd a Snowflake közvetlen másolatát. Ellenkező esetben használja a Snowflake beépített szakaszos másolatát.

Ha a Snowflake-ből szeretne adatokat másolni, az alábbi tulajdonságok támogatottak a Copy tevékenység forrásszakaszban.

Közvetlen másolat a Snowflake-ből

Ha a fogadó adattára és formátuma megfelel az ebben a szakaszban ismertetett feltételeknek, a Copy tevékenység használatával közvetlenül másolhatja a Snowflake-ből a fogadóba. A szolgáltatás ellenőrzi a beállításokat, és nem futtatja a Copy tevékenység, ha a következő feltételek nem teljesülnek:

• A fogadóhoz társított szolgáltatás az Azure Blob Storage megosztott hozzáférésű jogosultságkód-hitelesítéssel. Ha a következő támogatott formátumban szeretne közvetlenül adatokat másolni az Azure Data Lake Storage Gen2-be, létrehozhat egy Azure Blob-társított szolgáltatást SAS-hitelesítéssel az ADLS Gen2-fiókján, hogy elkerülje a Snowflake-ből származó szakaszos másolást.

• A fogadó adatformátuma parquet, tagolt szöveg vagy JSON, az alábbi konfigurációkkal:

  • Parquet formátum esetén a tömörítési kodek Nincs, Snappy vagy Lzo.
  • Tagolt szövegformátum esetén:
    • rowDelimiter \ r\n, vagy bármely karakter.
    • compressionnem lehet tömörítés, gzip, bzip2 vagy deflátum.
    • encodingNamealapértelmezettként marad, vagy utf-8 értékre van állítva.
    • quoteChar dupla idézőjel, egy idézőjel vagy üres sztring (nincs idézőjel karakter).
  • JSON-formátum esetén a közvetlen másolás csak azt az esetet támogatja, hogy a Snowflake-tábla vagy lekérdezés eredménye csak egyetlen oszlopból áll, és az oszlop adattípusa VARIANT, OBJECT vagy ARRAY.
    • compressionnem lehet tömörítés, gzip, bzip2 vagy deflátum.
    • encodingNamealapértelmezettként marad, vagy utf-8 értékre van állítva.
    • filePatterna másolási tevékenység fogadója alapértelmezés szerint marad, vagy a setOfObjects értékre van állítva.

• A másolási tevékenység forrásában additionalColumns nincs megadva.

• Nincs megadva oszlopleképezés.

Példa:

"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>"
            }
        }
    }
]

Szakaszos másolat a Snowflake-ből

Ha a fogadó adattára vagy formátuma natív módon nem kompatibilis a Snowflake COPY paranccsal, ahogyan azt az előző szakaszban említettük, engedélyezze a beépített szakaszos példányt egy köztes Azure Blob Storage-példány használatával. A szakaszos másolási funkció jobb átviteli sebességet is biztosít. A szolgáltatás adatokat exportál a Snowflake-ből az átmeneti tárolóba, majd átmásolja az adatokat a fogadóba, és végül megtisztítja az ideiglenes adatokat az átmeneti tárolóból. Az adatok előkészítéssel történő másolásával kapcsolatos részletekért tekintse meg a szakaszos másolást .

A funkció használatához hozzon létre egy Azure Blob Storage társított szolgáltatást , amely köztes előkészítésként hivatkozik az Azure Storage-fiókra. Ezután adja meg a enableStaging Copy tevékenység a tulajdonságokat és stagingSettings a tulajdonságokat.

Példa:

"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"
            }
        }
    }
]

Hópehely fogadóként

A Snowflake-összekötő a Snowflake COPY into [table] parancsát használja a legjobb teljesítmény eléréséhez. Támogatja az adatok írását a Snowflake-be az Azure-ban.

Ha a Snowflake COPY parancs natív módon támogatja a forrásadattárat és a formátumot, a Copy tevékenység közvetlenül másolhatja a forrásból a Snowflake-be. További részletekért lásd : A Snowflake közvetlen másolása. Ellenkező esetben használjon beépített szakaszos másolatot a Snowflake-hez.

Ha adatokat szeretne átmásolni a Snowflake-be, az alábbi tulajdonságok támogatottak a Copy tevékenység fogadó szakaszban.

Közvetlen másolat a Snowflake-be

Ha a forrásadattár és a formátum megfelel az ebben a szakaszban leírt feltételeknek, a Copy tevékenység használatával közvetlenül másolhat a forrásból a Snowflake-be. A szolgáltatás ellenőrzi a beállításokat, és nem futtatja a Copy tevékenység, ha a következő feltételek nem teljesülnek:

• A forráshoz társított szolgáltatás az Azure Blob Storage megosztott hozzáférésű jogosultságkód-hitelesítéssel. Ha a következő támogatott formátumban szeretne közvetlenül adatokat másolni az Azure Data Lake Storage Gen2-ből, létrehozhat egy Azure Blob-társított szolgáltatást SAS-hitelesítéssel az ADLS Gen2-fiókján, hogy elkerülje a Snowflake-be történő szakaszos másolást.

• A forrásadat formátuma parquet, tagolt szöveg vagy JSON az alábbi konfigurációkkal:

  • Parquet formátum esetén a tömörítési kodek Nincs vagy Snappy.

  • Tagolt szövegformátum esetén:

    • rowDelimiter \ r\n, vagy bármely karakter. Ha a sorelválasztó nem "\r\n", firstRowAsHeader akkor hamisnak kell lennie, és skipLineCount nincs megadva.
    • compressionnem lehet tömörítés, gzip, bzip2 vagy deflátum.
    • encodingName alapértelmezettként marad, vagy "UTF-8", "UTF-16", "UTF-16BE" értékre van állítva, "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", "WINDOWS-1255".
    • quoteChar dupla idézőjel, egy idézőjel vagy üres sztring (nincs idézőjel karakter).

  • JSON-formátum esetén a közvetlen másolás csak azt az esetet támogatja, hogy a fogadó Snowflake tábla csak egyetlen oszlopot tartalmaz, és az oszlop adattípusa VARIANT, OBJECT vagy ARRAY.

    • compressionnem lehet tömörítés, gzip, bzip2 vagy deflátum.
    • encodingNamealapértelmezettként marad, vagy utf-8 értékre van állítva.
    • Nincs megadva oszlopleképezés.

• A Copy tevékenység forrásban:

  • additionalColumns nincs megadva.
  • Ha a forrás egy mappa, recursive akkor igaz értékre van állítva.
  • prefix, modifiedDateTimeStart, modifiedDateTimeEndés enablePartitionDiscovery nincs megadva.

Példa:

"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"
                    }
                }
            }
        }
    }
]

Szakaszos másolat a Snowflake-be

Ha a forrásadattár vagy -formátum natív módon nem kompatibilis a Snowflake COPY paranccsal, ahogyan azt az előző szakaszban említettük, engedélyezze a beépített szakaszos példányt egy köztes Azure Blob Storage-példány használatával. A szakaszos másolási funkció jobb átviteli sebességet is biztosít. A szolgáltatás automatikusan átalakítja az adatokat, hogy megfeleljenek a Snowflake adatformátum-követelményeinek. Ezután meghívja a COPY parancsot az adatok Snowflake-be való betöltéséhez. Végül törli az ideiglenes adatokat a blobtárolóból. Az adatok átmeneti másolással történő másolásával kapcsolatos részletekért tekintse meg a szakaszos másolást .

A funkció használatához hozzon létre egy Azure Blob Storage társított szolgáltatást , amely köztes előkészítésként hivatkozik az Azure Storage-fiókra. Ezután adja meg a enableStaging Copy tevékenység a tulajdonságokat és stagingSettings a tulajdonságokat.

Példa:

"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"
            }
        }
    }
]

Adatfolyam-tulajdonságok leképezése

A leképezési adatfolyam adatainak átalakításakor a Snowflake-ban olvashat és írhat táblákba. További információkért tekintse meg a forrásátalakítást és a fogadóátalakítást a leképezési adatfolyamokban. Választhat, hogy a Snowflake-adatkészletet vagy egy beágyazott adatkészletet használ forrás- és fogadótípusként.

Forrásátalakítás

Az alábbi táblázat a Snowflake-forrás által támogatott tulajdonságokat sorolja fel. Ezeket a tulajdonságokat a Forrás beállításai lapon szerkesztheti. Az összekötő a Snowflake belső adatátvitelét használja.

Példák a Snowflake forrásszkriptjeire

Ha a Snowflake-adatkészletet használja forrástípusként, a társított adatfolyam-szkript a következő:

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

Beágyazott adatkészlet használata esetén a társított adatfolyam-szkript a következő:

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

Natív változáskövetés

Az Azure Data Factory mostantól támogatja a Snowflake egyik natív funkcióját, a változáskövetést, amely naplók formájában tartalmazza a változások nyomon követését. A hópehely ezen funkciója lehetővé teszi az adatok időbeli változásainak nyomon követését, ami hasznossá teszi az adatok növekményes betöltését és naplózását. A funkció használatához, amikor engedélyezi a Változás adatrögzítést, és kiválasztja a Snowflake Változáskövetést, létrehozunk egy Stream-objektumot a forrástáblához, amely lehetővé teszi a változáskövetést a forrás hópehelytáblán. Ezt követően a lekérdezésBEN a CHANGES záradékot használjuk, hogy csak az új vagy frissített adatokat kérje le a forrástáblából. Azt is javasoljuk, hogy ütemezze a folyamatokat, hogy a módosítások a snowflake forrástáblához beállított adatmegőrzési idő alatt legyenek felhasználva, máskülönben a felhasználó inkonzisztens viselkedést tapasztalhat a rögzített módosításokban.

Fogadó átalakítása

Az alábbi táblázat a Snowflake mosogató által támogatott tulajdonságokat sorolja fel. Ezeket a tulajdonságokat a Beállítások lapon szerkesztheti. Beágyazott adatkészlet használata esetén további beállítások jelennek meg, amelyek megegyeznek az adathalmaz tulajdonságainak szakaszában leírt tulajdonságokkal. Az összekötő a Snowflake belső adatátvitelét használja.

Snowflake fogadó szkript példák

Ha a Snowflake-adatkészletet fogadótípusként használja, a társított adatfolyam-szkript a következő:

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

Beágyazott adatkészlet használata esetén a társított adatfolyam-szkript a következő:

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

Lekérdezésleküldés optimalizálása

Ha a folyamat naplózási szintjét Nincs értékre állítja, kizárjuk a köztes átalakítási metrikák átvitelét, megakadályozva a Spark-optimalizálás lehetséges akadályait, és lehetővé tesszük a Snowflake által biztosított lekérdezésleküldési optimalizálást. Ez a leküldéses optimalizálás jelentős teljesítménybeli fejlesztéseket tesz lehetővé a nagyméretű, nagy adatkészletekkel rendelkező Snowflake-táblák esetében.

Keresési tevékenység tulajdonságai

A tulajdonságokról további információt a Keresési tevékenység című témakörben talál.

Kapcsolódó tartalom

A Copy tevékenység által forrásként és fogadóként támogatott adattárak listájáért tekintse meg a támogatott adattárakat és formátumokat.