使用 Azure Data Factory 或 Azure Synapse Analytics 在 Snowflake V1 中複製和轉換數據

適用於：Azure Data Factory Azure Synapse Analytics

本文概述如何使用 Azure Data Factory 和 Azure Synapse Analytics 管線中的複製活動，在 Snowflake 複製/貼上資料，並使用資料流程來轉換 Snowflake 中的資料。 若要深入了解，請閱讀 Data Factory 或 Azure Synapse Analytics 簡介文章。

支援的功能

此 Snowflake 連接器支援下列功能：

① 整合執行階段 ② 自我裝載整合執行階段

針對複製活動，此 Snowflake 連接器支援下列功能：

• 用 Snowflake 的 COPY into [location] (英文) 命令從 Snowflake 複製資料，藉此達到最佳效能。
• 用 Snowflake 的 COPY into [table] (英文) 命令將資料複製到 Snowflake，藉此達到最佳效能。 這項命令支援 Azure 上的 Snowflake。
• 如果需要 Proxy 從自我裝載整合執行階段連線到 Snowflake，您必須在整合執行階段主機上設定 HTTP_PROXY 和 HTTPS_PROXY 環境變數。

必要條件

如果您的資料存放區位於內部部署網路、Azure 虛擬網路或 Amazon 虛擬私人雲端中，則必須設定自我裝載整合執行階段以與其連線。 請務必將自我裝載整合執行階段使用的 IP 位址新增至允許清單。

如果您的資料存放區是受控雲端資料服務，則可使用 Azure Integration Runtime。 如果只能存取防火牆規則中核准的 IP，您可以將 Azure Integration Runtime IP 新增至允許清單。

用於來源或接收的 Snowflake 帳戶應該具有資料庫的必要 USAGE 存取權，以及結構描述及其下的資料表/檢視表的讀取/寫入存取權。 此外，其也應該在結構描述上具有 CREATE STAGE，才能使用 SAS URI 建立外部階段。

必須設定下列帳戶屬性值

如需 Data Factory 支援的網路安全性機制和選項的詳細資訊，請參閱資料存取策略。

開始使用

若要透過管線執行複製活動，您可以使用下列其中一個工具或 SDK：

• 資料複製工具
• Azure 入口網站
• .NET SDK
• Python SDK
• Azure PowerShell
• REST API
• Azure Resource Manager 範本

使用 UI 建立 Snowflake 的連結服務

使用下列步驟，在 Azure 入口網站 UI 中建立 Snowflake 的連結服務。

1. 前往 Azure Data Factory 或 Synapse 工作區的 [管理] 索引標籤，選取 [連結服務]，然後按一下 [新增]：

   • Azure Data Factory
   • Azure Synapse

   

2. 搜尋 Snowflake 並選取 Snowflake 連接器。

   

3. 設定服務詳細資料，測試連線，然後建立新的連結服務。

   

連接器設定詳細資料

下列各節提供屬性的相關詳細資料，這些屬性會定義 Snowflake 連接器的特定實體。

連結服務屬性

Snowflake 連接器支援下列驗證類型。 如需詳細資料，請參閱對應章節。

• 基本驗證

基本驗證

使用 Basic 身分驗證時，Snowflake 連結服務支援下列屬性。

範例：

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

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

資料集屬性

如需可用來定義資料集的區段和屬性完整清單，請參閱資料集一文。

以下是針對 Snowflake 資料集支援的屬性。

範例：

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

複製活動屬性

如需可用來定義活動的區段和屬性完整清單，請參閱管線一文。 本節提供 Snowflake 來源和接收器支援的屬性清單。

以 Snowflake 做為來源

Snowflake 連接器會用 Snowflake 的 COPY into [location] (英文) 命令達到最佳效能。

如果 Snowflake COPY 命令原生支援接收器資料存放區和格式，則可使用複製活動，直接從 Snowflake 複製到接收器。 如需詳細資訊，請參閱從 Snowflake 直接複製。 若否，請使用內建的從 Snowflake 暫存複本。

若要從 Snowflake 複製資料，複製活動的 [source] 區段支援下列屬性。

從 Snowflake 直接複製

如果您的接收器資料存放區和格式符合本節所述準則，則可使用複製活動，直接從 Snowflake 複製到接收器。 該服務會檢查設定，並在不符合下列準則時讓複製活動執行失敗：

• 接收器連結服務是具有共用存取簽章驗證的 Azure Blob 儲存體。 如果您想要以下列支援的格式直接將資料複製到 Azure Data Lake Storage Gen2，請針對 ADLS Gen2 帳戶建立具有 SAS 驗證的 Azure Blob 連結服務，以避免使用從 Snowflake 暫存複本。

• 接收器資料格式包括 Parquet、分隔符號文字 或 JSON，並具有下列設定：

  • 針對 Parquet 格式，壓縮轉碼器為 None、Snappy 或 Lzo。
  • 針對分隔符號文字格式：
    • rowDelimiter 為 \r\n 或任何單一字元。
    • compression 可以是無壓縮、gzip、bzip2 或 deflate。
    • encodingName 會保留為預設值，或設定為 utf-8。
    • quoteChar 為雙引號、單引號或空字串 (沒有引號字元)。
  • 針對 JSON 格式，直接複製僅支援來源 Snowflake 資料表或查詢結果只有單一資料行的案例，且此資料行的資料類型為 VARIANT、OBJECT 或 ARRAY。
    • compression 可以是無壓縮、gzip、bzip2 或 deflate。
    • encodingName 會保留為預設值，或設定為 utf-8。
    • filePattern 在複製活動接收器中保留為預設值，或設定為 setOfObjects。

• 在複製活動來源中，不會指定 additionalColumns。

• 不會指定資料行對應。

範例：

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

從 Snowflake 暫存複本

當您的接收器資料存放區或格式與 Snowflake COPY 命令原生不相容時，如上一節所述，請使用過渡 Azure Blob 儲存體執行個體啟用內建分段複製。 分段複製功能也能提供更好的輸送量。 服務會將資料從 Snowflake 匯出至暫存儲存體，然後將資料複製到接收器，最後從暫存儲存體中清除暫存資料。 如需使用暫存複製資料的詳細資料，請參閱暫存複製。

如要使用此功能，請建立 Azure Blob 儲存體連結服務，參考具有過渡暫存的 Azure 儲存體帳戶。 然後，在複製活動中指定 enableStaging 和 stagingSettings 屬性。

範例：

"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 做為接收器

Snowflake 連接器會用 Snowflake 的 COPY into [table] (英文) 命令達到最佳效能。 此功能支援將資料寫入 Azure 上的 Snowflake。

如果 Snowflake COPY 命令原生支援來源資料存放區和格式，則可使用複製活動，直接從來源複製到 Snowflake。 如需詳細資訊，請參閱直接複製到 Snowflake。 若否，請使用內建的暫存複本到 Snowflake。

若要將資料複製到 Snowflake，複製活動的 [sink] 區段支援下列屬性。

直接複製到 Snowflake

如果您的來源資料存放區和格式符合本節所述的準則，則可使用複製活動，直接從來源複製到 Snowflake。 該服務會檢查設定，並在不符合下列準則時讓複製活動執行失敗：

• 來源連結服務是具有共用存取簽章驗證的 Azure Blob 儲存體。 如果您想要以下列支援的格式直接從 Azure Data Lake Storage Gen2 複製資料，請針對 ADLS Gen2 帳戶建立具有 SAS 驗證的 Azure Blob 連結服務，以避免使用暫存複本到 Snowflake。

• 來源資料格式包括 Parquet、分隔符號文字 或 JSON，並具有下列設定：

  • 針對 Parquet 格式，壓縮轉碼器為 None 或 Snappy。

  • 針對分隔符號文字格式：

    • rowDelimiter 為 \r\n 或任何單一字元。 如果資料列分隔符號不是 “\r\n”，則 firstRowAsHeader 必須為 false，且不會指定 skipLineCount。
    • compression 可以是無壓縮、gzip、bzip2 或 deflate。
    • encodingName 保留為預設值，或設定為「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-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 為雙引號、單引號或空字串 (沒有引號字元)。

  • 針對 JSON 格式，直接複製僅支援接收器 Snowflake 資料表只有單一資料行的案例，且此資料行的資料類型為 VARIANT、OBJECT 或 ARRAY。

    • compression 可以是無壓縮、gzip、bzip2 或 deflate。
    • encodingName 會保留為預設值，或設定為 utf-8。
    • 不會指定資料行對應。

• 在複製活動來源中：

  • 不會指定 additionalColumns。
  • 如果來源是資料夾，recursive 會設定為 true。
  • 不會指定 prefix、modifiedDateTimeStart、modifiedDateTimeEnd 和 enablePartitionDiscovery。

範例：

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

暫存複本到 Snowflake

您的來源資料存放區或格式與 Snowflake COPY 命令原生不相容時，如上一節所述，請使用過渡 Azure Blob 儲存體執行個體啟用內建分段複製。 分段複製功能也能提供更好的輸送量。 該服務會自動轉換資料，以符合 Snowflake 的資料格式需求。 之後系統會叫用 COPY 命令，將資料載入 Snowflake。 最後，它會清除 Blob 儲存體中的暫存資料。 如需使用暫存複製資料的詳細資料，請參閱暫存複製。

如要使用此功能，請建立 Azure Blob 儲存體連結服務，參考具有過渡暫存的 Azure 儲存體帳戶。 然後，在複製活動中指定 enableStaging 和 stagingSettings 屬性。

範例：

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

對應資料流程屬性

在對應資料流中轉換資料時，您可以從 Snowflake 資料表中讀取並寫入資料。 如需詳細資訊，請參閱對應資料流程中的來源轉換和接收轉換。 您可以選擇使用 Snowflake 資料集或內嵌資料集做為來源和接收器型別。

來源轉換

下表列出 Snowflake 來源支援的屬性。 您可以在 [來源選項] 索引標籤中編輯這些屬性。連接器會使用 Snowflake 的內部資料傳輸 (英文)。

Snowflake 來源指令碼範例

使用 Snowflake 資料集做為來源型別時，相關聯的資料流程指令碼為：

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

如果您使用內嵌資料集，相關聯的資料流程指令碼為：

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

原生變更追蹤

Azure Data Factory 現在於 Snowflake 中支援稱為變更追蹤的原生功能，其中包含以記錄形式追蹤變更。 Snowflake 的這項功能可讓我們追蹤一段時間的資料變更，使其適用於累加資料載入和稽核用途。 若要利用這項功能，當您啟用異動資料擷取並選取 Snowflake 變更追蹤時，我們會為來源資料表建立 Stream 物件，以啟用來源 Snowflake 資料表上的變更追蹤。 接著，我們會在查詢中使用 CHANGES 子句，只從來源資料表擷取新的或更新的資料。 此外，建議排程管線，讓變更在為 snowflake 來源資料表設定的資料保留時間間隔內耗用，否則使用者可能會在擷取的變更中看到不一致的行為。

接收轉換

下表列出 Snowflake 接收器支援的屬性。 您可以在 [設定] 索引標籤中編輯這些屬性。使用內嵌資料集時，您將會看到其他設定，這些設定與資料集屬性一節所述的屬性相同。 連接器會使用 Snowflake 的內部資料傳輸 (英文)。

Snowflake 接收器指令碼範例

使用 Snowflake 資料集做為接收器型別時，相關聯的資料流程指令碼為：

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

如果您使用內嵌資料集，相關聯的資料流程指令碼為：

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

查詢下推最佳化

藉由將管線記錄層級設定為「無」，我們會排除中繼轉換計量的傳輸、防止 Spark 最佳化的潛在障礙，以及啟用 Snowflake 所提供的查詢下推最佳化。 此下推最佳化可讓具有廣泛資料集的大型 Snowflake 資料表大幅提升效能。

查閱活動屬性

如需屬性的詳細資訊，請參閱查閱活動。

相關內容

如需複製活動支援作為來源和接收器的資料存放區清單，請參閱支援的資料存放區和格式。