Копирование и преобразование данных в Snowflake версии 2 с помощью Фабрика данных Azure или Azure Synapse Analytics

ПРИМЕНЯЕТСЯ К: Azure Data Factory Azure Synapse Analytics

Tip

Попробуйте Data Factory в Microsoft Fabric — комплексное аналитическое решение для предприятий. Microsoft Fabric охватывает все, от перемещения данных до обработки и анализа данных в режиме реального времени, бизнес-аналитики и отчетности. Узнайте, как бесплатно запустить новую пробную версию !

В этой статье описано использование действия Copy в Фабрике данных Azure и конвейерах Azure Synapse, чтобы копировать данные в хранилище Snowflake и из него, а также использовать Поток данных для преобразования данных в Snowflake. Дополнительные сведения см. в вводной статье о Data Factory или Azure Synapse Analytics.

Important

Соединитель Snowflake версии 1 находится на этапе удаления. Рекомендуется обновить соединитель Snowflake с версии 1 до версии 2.

Поддерживаемые возможности

Этот соединитель Snowflake поддерживается для возможностей, указанных ниже:

Поддерживаемые возможности	IR
Действие копирования (источник/приемник)	(1) (2)
Сопоставление потока данных (источник или приемник)	①
Действие подстановки	(1) (2)
Действие скрипта (применение версии 1.1 при использовании параметра скрипта)	(1) (2)

(1) Среда выполнения интеграции Azure (2) Локальная среда выполнения интеграции

Для действия Copy данный соединитель Snowflake поддерживает функции, описанные ниже.

• Скопируйте данные из Snowflake, используя команду COPY into [location], чтобы добиться оптимальной производительности.
• Скопируйте данные в Snowflake, который использует команду COPY INTO [table] для максимальной производительности. Поддерживает Snowflake в Azure.
• Если для подключения к Snowflake из автономной среды выполнения интеграции требуется прокси-сервер, необходимо настроить переменные среды для HTTP_PROXY и HTTPS_PROXY на узле среды выполнения интеграции.

Prerequisites

Если хранилище данных находится в локальной сети, виртуальной сети Azure или Amazon Virtual Private Cloud, необходимо настроить локальную среду выполнения интеграции для подключения к ней. Обязательно добавьте IP-адреса, которые использует локальная среда выполнения интеграции, в список разрешенных.

Если же хранилище данных представляет собой управляемую облачную службу данных, можно использовать Azure Integration Runtime. Если доступ ограничен IP-адресами, утвержденными в правилах брандмауэра, можно добавить IP-адреса среды выполнения интеграции Azure в список разрешенных адресов.

Учетная запись Snowflake, используемая для источника или приемника, должна иметь необходимый USAGE доступ к базе данных и доступ на чтение и запись в схеме и таблицах и представлениях под ним. Кроме того, у нее должны быть права CREATE STAGE для схемы, чтобы создавать внешний этап с URI SAS.

Необходимо задать приведенные ниже значения свойств учетной записи.

Property	Description	Required	Default
REQUIRE_STORAGE_INTEGRATION_FOR_STAGE_CREATION	Указывает, требуется ли объект интеграции хранилища в качестве облачных учетных данных при создании именованного внешнего этапа (с помощью CREATE STAGE) для доступа к расположению частного облака.	FALSE	FALSE
REQUIRE_STORAGE_INTEGRATION_FOR_STAGE_OPERATION	Указывает, требуется ли использовать именованный внешний этап, который ссылается на объект интеграции хранилища в качестве облачных учетных данных при загрузке данных из частного облачного хранилища или выгрузке данных в него.	FALSE	FALSE

Дополнительные сведения о механизмах и вариантах безопасности сети, поддерживаемых фабрикой данных, см. в стратегиях доступа к данным.

Get started

Для выполнения действия копирования с конвейером можно использовать один из следующих средств или пакетов SDK:

• Средство копирования данных
• Портал Azure
• Пакет SDK для .NET
• Пакет SDK для Python
• Azure PowerShell
• REST API
• Шаблон Azure Resource Manager

Создание связанной службы со Snowflake с помощью пользовательского интерфейса

Выполните следующие действия, чтобы создать связанную службу для Snowflake с помощью пользовательского интерфейса на портале Azure.

1. Перейдите на вкладку "Управление" в рабочей области Фабрики данных Azure или Synapse и выберите "Связанные службы", после чего нажмите "Создать":

   Фабрика данных Azure

   

   Azure Synapse

   

2. Выполните поиск Snowflake и выберите соединитель Snowflake.

   

3. Настройте сведения о службе, проверьте подключение и создайте связанную службу.

   

Сведения о конфигурации соединителя

Следующие разделы содержат сведения о свойствах, используемых для определения сущностей, относящихся к соединителю Snowflake.

Свойства связанной службы

Эти универсальные свойства поддерживаются для связанной службы Snowflake:

Property	Description	Required
type	Свойство type должно иметь значение SnowflakeV2.	Yes
version	Версия, которую вы указали. Рекомендуется обновить до последней версии, чтобы воспользоваться новыми улучшениями.	Да для версии 1.1
accountIdentifier	Имя учетной записи вместе с ее организацией. Например, myorg-account123.	Yes
database	База данных по умолчанию, используемая для сеанса после подключения.	Yes
warehouse	Виртуальное хранилище по умолчанию, используемое для сеанса после подключения.	Yes
authenticationType	Тип проверки подлинности, используемый для подключения к службе Snowflake. Допустимые значения: "Базовый " (по умолчанию) и KeyPair. Обратитесь к соответствующим разделам ниже, в которых описываются дополнительные свойства и примеры.	No
role	Роль безопасности по умолчанию, используемая для сеанса после подключения.	No
host	Имя узла учетной записи Snowflake. Например: contoso.snowflakecomputing.com. .cn также поддерживается.	No
connectVia	Среда выполнения интеграции , используемая для подключения к хранилищу данных. Можно использовать среду выполнения интеграции Azure или локальную среду выполнения интеграции (если хранилище данных расположено в частной сети). Если не указано другое, по умолчанию используется среда выполнения интеграции Azure.	No

В зависимости от вашего случая можно задать следующие дополнительные свойства подключения в связанной службе.

Property	Description	Required	Значение по умолчанию
UseUtcTimestamps	Укажите false, чтобы возвращать тип TIMESTAMP_LTZ и тип TIMESTAMP_TZ в корректной временной зоне, а также тип TIMESTAMP_NTZ без информации о временной зоне. Установите значение true, чтобы возвращать все типы временных меток Snowflake в формате UTC.	No	false
schema	Указывает схему сеанса запроса после подключения.	No	/

Этот соединитель Snowflake поддерживает следующие типы проверки подлинности. Дополнительные сведения см. в соответствующих разделах.

• Обычная проверка подлинности
• Проверка подлинности пары ключей

Basic authentication

Чтобы использовать обычную проверку подлинности, в дополнение к универсальным свойствам, описанным в предыдущем разделе, укажите следующие свойства:

Property	Description	Required
user	Имя входа для пользователя Snowflake.	Yes
password	Пароль для пользователя Snowflake. Пометьте это поле как тип SecureString для безопасного хранения. Вы также можете ссылаться на секрет, хранящийся в Azure Key Vault.	Yes

Example:

{
    "name": "SnowflakeV2LinkedService",
    "properties": {
        "type": "SnowflakeV2",
        "typeProperties": {
            "accountIdentifier": "<accountIdentifier>",
            "database": "<database>",
            "warehouse": "<warehouse>",
            "authenticationType": "Basic",
            "user": "<username>",
            "password": {
                "type": "SecureString",
                "value": "<password>"
            },
            "role": "<role>"
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

Пароль в Azure Key Vault:

{
    "name": "SnowflakeV2LinkedService",
    "properties": {
        "type": "SnowflakeV2",
        "typeProperties": {
            "accountIdentifier": "<accountIdentifier>",
            "database": "<database>",
            "warehouse": "<warehouse>",
            "authenticationType": "Basic",
            "user": "<username>",
            "password": {
                "type": "AzureKeyVaultSecret",
                "store": { 
                    "referenceName": "<Azure Key Vault linked service name>",
                    "type": "LinkedServiceReference"
                }, 
                "secretName": "<secretName>"
            }
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

Проверка подлинности пары ключей

Чтобы использовать аутентификацию с помощью пары ключей, необходимо настроить и создать пользователя с аутентификацией по паре ключей в Snowflake, ссылаясь на Аутентификация по паре ключей и ротация ключей. Затем запишите закрытый ключ и парольную фразу (необязательно), которая используется для определения связанной службы.

В дополнение к общим свойствам, описанных в предыдущих разделах, укажите следующие свойства:

Property	Description	Required
user	Имя входа для пользователя Snowflake.	Yes
privateKey	Закрытый ключ, используемый для проверки подлинности пары ключей. Чтобы обеспечить допустимость закрытого ключа при отправке в Фабрика данных Azure, и учитывая, что файл privateKey содержит новые символы (\n), важно правильно отформатировать содержимое privateKey в строковой литеральной форме. Этот процесс включает добавление \n явным образом в каждую новую линию.	Yes
privateKeyPassphrase	Парольная фраза, используемая для расшифровки закрытого ключа, если она зашифрована.	No

Example:

{
    "name": "SnowflakeV2LinkedService",
    "properties": {
        "type": "SnowflakeV2",
        "typeProperties": {
            "accountIdentifier": "<accountIdentifier>",
            "database": "<database>",
            "warehouse": "<warehouse>",
            "authenticationType": "KeyPair",
            "user": "<username>",
            "privateKey": {
                "type": "SecureString",
                "value": "<privateKey>"
            },
            "privateKeyPassphrase": { 
                "type": "SecureString",
                "value": "<privateKeyPassphrase>"
            },
            "role": "<role>"
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

Note

Для сопоставления потоков данных рекомендуется создать новый закрытый ключ RSA с помощью стандарта PKCS#8 в формате PEM (P8-файл).

Свойства набора данных

Полный список разделов и свойств, доступных для определения наборов данных, см. в статье " Наборы данных ".

Для набора данных Snowflake поддерживаются свойства, указанные ниже.

Property	Description	Required
type	Свойство типа набора данных должно иметь значение SnowflakeV2Table.	Yes
schema	Имя схемы. Обратите внимание, что имя схемы чувствительно к регистру.	"Нет" для источника, "Да" для приемника
table	Имя таблицы или представления. Обратите внимание, что имя таблицы чувствительно к регистру.	"Нет" для источника, "Да" для приемника

Example:

{
    "name": "SnowflakeV2Dataset",
    "properties": {
        "type": "SnowflakeV2Table",
        "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 [расположение], чтобы обеспечить лучшую производительность.

Если хранилище данных приемника и формат изначально поддерживаются командой COPY из Snowflake, можно использовать действие Copy для прямого копирования из Snowflake в приемник. Для получения подробностей см. раздел "Прямая копия из Snowflake". В противном случае используйте встроенную функцию «Staged copy from Snowflake».

Чтобы скопировать данные из Snowflake, в разделе источника действия копирования поддерживаются следующие свойства.

Property	Description	Required
type	Свойство type источника действия Copy должно иметь значение SnowflakeV2Source.	Yes
query	Определяет SQL-запрос для чтения данных из Snowflake. Если имена схемы, таблицы и столбцов содержат строчные буквы, заключайте идентификатор объекта в кавычки, например select * from "schema"."myTable". Выполнение хранимой процедуры не поддерживается.	No
exportSettings	Дополнительные параметры, используемые для получения данных из Snowflake. Можно настроить параметры, поддерживаемые командой COPY, в команде, которую будет передавать служба при вызове оператора.	Yes
treatDecimalAsString	Укажите, чтобы обрабатывать десятичный тип как строковый в деятельности поиска и скрипта. Значение по умолчанию — false. Это свойство поддерживается только в версии 1.1.	No
В разделе exportSettings:
type	Тип команды экспорта, для которой задано значение SnowflakeExportCopyCommand.	Yes
storageIntegration	Укажите имя интеграции хранилища, созданной в Snowflake. Инструкции по использованию интеграции хранилища см. в разделе "Настройка интеграции хранилища Snowflake".	No
additionalCopyOptions	Дополнительные параметры копирования, предоставляемые в виде словаря для пар "ключ-значение". Примеры: MAX_FILE_SIZE, OVERWRITE. Дополнительные сведения см. в разделе "Параметры копирования Snowflake".	No
additionalFormatOptions	Дополнительные параметры формата файла, предоставляемые для команды КОПИРОВАТЬ в виде словаря для пар "ключ-значение". Примеры: DATE_FORMAT, TIME_FORMAT, TIMESTAMP_FORMAT, NULL_IF. Дополнительные сведения см. в разделе "Параметры типа формата Snowflake". При использовании NULL_IF значение NULL в Snowflake преобразуется в указанное значение (которое должно быть одно кавычек) при записи в текстовый файл с разделителями в промежуточном хранилище. Это указанное значение обрабатывается как NULL при чтении из промежуточного файла в хранилище приемника. Значение по умолчанию — 'NULL'.	No

Note

Убедитесь, что у вас есть разрешение на выполнение следующей команды и доступ к схеме INFORMATION_SCHEMA и столбцам таблицы.

• COPY INTO <location>

Прямое копирование из Snowflake

Если хранилище данных приемника и формат соответствуют критериям, описанным в этом разделе, действие Copy можно использовать для прямого копирования из Snowflake в приемник. Служба проверяет параметры и завершает выполнение действие Copy, если следующие критерии не выполнены:

• При указании storageIntegration в источнике:

  Хранилище данных приемника — это Хранилище BLOB-объектов Azure, который вы указали во внешней стадии в Snowflake. Перед копированием данных необходимо выполнить следующие действия:

  1. Создайте связанную службу Azure Blob Storage для приемного хранилища Azure Blob Storage с любыми поддерживаемыми типами проверки подлинности.

  2. Предоставьте по крайней мере роль участника данных хранилища BLOB-объектов сервисному принципалу Snowflake в управлении доступом (IAM) к хранилищу Azure Blob Storage .

• Если вы не указываете storageIntegration в источнике:

  Связанная служба приемника — это хранилище BLOB-объектов Azure с проверкой подлинности по общей подписи доступа. Если вы хотите напрямую скопировать данные в Azure Data Lake Storage 2-го поколения в следующем поддерживаемом формате, можно создать связанную службу Azure Blob Storage с проверкой подлинности SAS для вашей учетной записи Azure Data Lake Storage 2-го поколения, чтобы избежать использования промежуточной копии из 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 действия копирования не указано.

• Сопоставление столбцов не указано.

Example:

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

Промежуточное копирование из Snowflake

Если хранилище данных приемника или формат не совместимы с командой Snowflake COPY, как упоминалось в последнем разделе, включите встроенную поэтапное копирование с помощью промежуточного экземпляра хранилища BLOB-объектов Azure. Функция поэтапного копирования также обеспечивает более высокую пропускную способность. Служба экспортирует данные из Snowflake в промежуточное хранилище, затем копирует данные в приемник и, наконец, очищает промежуточное хранилище от временных данных. Сведения о копировании данных с помощью промежуточного копирования см. в разделе "Поэтапное копирование ".

Чтобы использовать эту функцию, создайте связанную службу хранилища BLOB-объектов Azure, которая ссылается на учетную запись хранения Azure в качестве временной промежуточной. Затем укажите свойства enableStaging и stagingSettings в действии Copy.

• При указании storageIntegration в источнике промежуточный промежуток Хранилище BLOB-объектов Azure должен быть тем, который вы указали во внешней стадии в Snowflake. Убедитесь, что для нее создается связанная служба хранилища BLOB-объектов Azure с любой поддерживаемой проверкой подлинности при использовании среды выполнения интеграции Azure или с анонимным ключом учетной записи, подписанным URL-адресом или проверкой подлинности субъекта-службы при использовании локальной среды выполнения интеграции. Кроме того, предоставьте основному представителю службы Snowflake по крайней мере роль участника данных хранилища Blob в управлении доступом (IAM) к промежуточному хранилищу Blob в Azure.

• Если вы не указываете storageIntegration в источнике, промежуточная Хранилище BLOB-объектов Azure связанная служба должна использовать проверку подлинности подписи общего доступа, как требуется команде Snowflake COPY. Убедитесь, что вы предоставляете надлежащее разрешение на доступ к Snowflake в промежуточном Хранилище BLOB-объектов Azure. Дополнительные сведения об этом см. в этой статье.

Example:

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

При выполнении поэтапной копии из Snowflake важно задать поведение копирования приемника для слияния файлов. Этот параметр гарантирует правильность обработки и объединения всех секционированных файлов, предотвращая проблему, из-за которой копируется только последний секционированный файл.

Пример конфигурации

{
    "type": "Copy",
    "source": {
        "type": "SnowflakeSource",
        "query": "SELECT * FROM my_table"
    },
    "sink": {
        "type": "AzureBlobStorage",
        "copyBehavior": "MergeFiles"
    }
}

Note

Если не задать поведение копирования приемника как слияние файлов, это может привести к копированию только последнего секционированного файла.

Snowflake в качестве приемника

Соединитель Snowflake использует команду Snowflake COPY into [table], чтобы добиться оптимальной производительности. Он поддерживает запись данных в Snowflake в Azure.

Если хранилище данных источника и формат изначально поддерживаются командой КОПИРОВАТЬ из Snowflake, можно использовать действие копирования для прямого копирования из источника в Snowflake. Дополнительные сведения см. в разделе "Прямая копия в Snowflake". В противном случае используйте встроенную функцию Staged Copy to Snowflake.

Чтобы скопировать данные в Snowflake, в секции приемника копирования поддерживаются следующие свойства.

Property	Description	Required
type	Свойство type приемника операции копирования, установленное на SnowflakeV2Sink.	Yes
preCopyScript	Перед записью данных в базу данных Snowflake при каждом запуске указывайте SQL-запрос для выполнения действия Copy. Это свойство используется для очистки предварительно загруженных данных.	No
importSettings	Дополнительные параметры, используемые для записи данных в Snowflake. Можно настроить параметры, поддерживаемые командой COPY, в команде, которую будет передавать служба при вызове оператора.	Yes
В разделе importSettings:
type	Тип команды импорта, для которой задано значение SnowflakeImportCopyCommand.	Yes
storageIntegration	Укажите имя интеграции хранилища, созданной в Snowflake. Инструкции по использованию интеграции хранилища см. в разделе "Настройка интеграции хранилища Snowflake".	No
additionalCopyOptions	Дополнительные параметры копирования, предоставляемые в виде словаря для пар "ключ-значение". Примеры: ON_ERROR, FORCE, LOAD_UNCERTAIN_FILES. Дополнительные сведения см. в разделе "Параметры копирования Snowflake".	No
additionalFormatOptions	Дополнительные параметры формата файла, предоставляемые для команды КОПИРОВАТЬ, предоставляются в виде словаря для пар "ключ-значение". Примеры: DATE_FORMAT, TIME_FORMAT, TIMESTAMP_FORMAT. Дополнительные сведения см. в разделе "Параметры типа формата Snowflake".	No

Note

Убедитесь, что у вас есть разрешение на выполнение следующей команды и доступ к схеме INFORMATION_SCHEMA и столбцам таблицы.

• SELECT CURRENT_REGION()
• COPY INTO <table>
• SHOW REGIONS
• CREATE OR REPLACE STAGE
• DROP STAGE

Прямое копирование в Snowflake

Если хранилище данных источника и формат соответствуют критериям, описанным в этом разделе, действие Copy можно использовать для прямого копирования из источника в Snowflake. Служба проверяет параметры и завершает выполнение действие Copy, если следующие критерии не выполнены:

• При указании storageIntegration в приемнике:

  Исходное хранилище данных — это Хранилище BLOB-объектов Azure, которые вы указали на внешнем этапе в Snowflake. Перед копированием данных необходимо выполнить следующие действия:

  1. Создайте связанную службу для исходного хранилища Blob в Azure, используя любые поддерживаемые типы аутентификации.

  2. Предоставьте по крайней мере роль Storage Blob Data Reader субъекту-службе Snowflake в исходной системе управления доступом хранилища BLOB-объектов Azure (IAM).

• Если не указать storageIntegration в приемнике:

  Связанная служба — это хранилище BLOB-объектов Azure с проверкой подлинности с использованием общей подписи доступа. Если вы хотите напрямую скопировать данные из Azure Data Lake Storage Gen2 в одном из следующих поддерживаемых форматов, можно создать связанную службу BLOB-хранилища Azure с аутентификацией SAS для вашей учетной записи Azure Data Lake Storage Gen2, чтобы избежать использования поэтапной копии в 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-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.
    • Сопоставление столбцов не указано.

• В источнике действия Copy:

  • additionalColumns не указан.
  • Если источником является папка, для recursive установлено значение true.
  • prefix, modifiedDateTimeStart, modifiedDateTimeEnd и enablePartitionDiscovery не указаны.

Example:

"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": "SnowflakeV2Sink",
                "importSettings": {
                    "type": "SnowflakeImportCopyCommand",
                    "copyOptions": {
                        "FORCE": "TRUE",
                        "ON_ERROR": "SKIP_FILE"
                    },
                    "fileFormatOptions": {
                        "DATE_FORMAT": "YYYY-MM-DD"
                    },
                    "storageIntegration": "< Snowflake storage integration name >"
                }
            }
        }
    }
]

Промежуточное копирование в Snowflake

Если исходное хранилище данных или формат не совместимы с командой Snowflake COPY, как упоминалось в последнем разделе, включите встроенную поэтапное копирование с помощью промежуточного экземпляра хранилища BLOB-объектов Azure. Функция поэтапного копирования также обеспечивает более высокую пропускную способность. Служба автоматически преобразует данные, чтобы они соответствовали требованиям к формату данных Snowflake. Затем она вызывает команду КОПИРОВАТЬ для загрузки данных в Snowflake. Наконец, она очищает ваши временные данные из хранилища BLOB-объектов. Сведения о копировании данных с помощью промежуточного копирования см. в разделе "Поэтапное копирование ".

Чтобы использовать эту функцию, создайте связанную службу хранилища BLOB-объектов Azure, которая ссылается на учетную запись хранения Azure в качестве временной промежуточной. Затем укажите свойства enableStaging и stagingSettings в действии Copy.

• При указании storageIntegration в приемнике промежуточный промежуточный Хранилище BLOB-объектов Azure должен быть тем, который вы указали во внешней стадии в Snowflake. Убедитесь, что для нее создается связанная служба хранилища BLOB-объектов Azure с любой поддерживаемой проверкой подлинности при использовании среды выполнения интеграции Azure или с анонимным ключом учетной записи, подписанным URL-адресом или проверкой подлинности субъекта-службы при использовании локальной среды выполнения интеграции. Кроме того, предоставьте по крайней мере роль Storage Blob Data Reader субъекту-службе Snowflake в системе управления доступом (IAM) промежуточного хранилища Azure Blob.

• Если вы не указываете storageIntegration в приемнике, промежуточной Хранилище BLOB-объектов Azure связанной службе необходимо использовать проверку подлинности подписи общего доступа в соответствии с командой Snowflake COPY.

Example:

"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": "SnowflakeV2Sink",
                "importSettings": {
                    "type": "SnowflakeImportCopyCommand",
                    "storageIntegration": "< Snowflake storage integration name >"
                }
            },
            "enableStaging": true,
            "stagingSettings": {
                "linkedServiceName": {
                    "referenceName": "MyStagingBlob",
                    "type": "LinkedServiceReference"
                },
                "path": "mystagingpath"
            }
        }
    }
]

Свойства потока данных для сопоставления

При преобразовании данных в потоке данных для сопоставления можно выполнять операции чтения и записи в таблицах в Snowflake. Дополнительные сведения см. в преобразовании источника и преобразовании приемника в потоках сопоставления данных. Вы можете использовать набор данных Snowflake или встроенный набор данных в качестве источника и типа приемника.

Преобразование источника

В таблице, приведенной ниже, указаны свойства, поддерживаемые источником Snowflake. Эти свойства можно изменить на вкладке "Параметры источника ". Соединитель использует внутреннюю передачу данных Snowflake.

Name	Description	Required	Допустимые значения	Свойство скрипта для потока данных
Table	Если в качестве входных данных выбрать вариант "Таблица", поток данных будет получать все данные из таблицы, указанной в наборе данных Snowflake, или в параметрах источника при использовании встроенного набора данных.	No	String	(только для встроенного набора данных) tableName schemaName
Query	Если в качестве входных данных выбрать "Запрос", введите запрос для получения данных из Snowflake. Этот параметр переопределяет любую таблицу, выбранную в наборе данных. Если имена схемы, таблицы и столбцов содержат строчные буквы, заключайте идентификатор объекта в кавычки, например select * from "schema"."myTable".	No	String	query
Включение добавочного извлечения (предварительная версия)	Используйте этот параметр, чтобы сообщить ADF обработать только строки, которые изменились с момента последнего выполнения конвейера.	No	Boolean	enableCdc
Добавочный столбец	При использовании функции добавочного извлечения необходимо выбрать столбец даты и времени и числового столбца, который вы хотите использовать в качестве водяного знака в исходной таблице.	No	String	waterMarkColumn
Включение snowflake Отслеживание изменений (предварительная версия)	Этот параметр позволяет ADF использовать технологию отслеживания измененных данных Snowflake для обработки только разностных данных с момента выполнения предыдущего конвейера. Этот параметр автоматически загружает разностные данные с операциями вставки строк, обновления и удаления без необходимости добавлять добавочный столбец.	No	Boolean	enableNativeCdc
Чистые изменения	При использовании отслеживания изменений snowflake можно использовать этот параметр, чтобы получить дедупликированные измененные строки или исчерпывающие изменения. Дедупликированные измененные строки будут отображать только последние версии строк, которые изменились с заданной точки во времени, в то время как исчерпывающие изменения будут отображать все версии каждой строки, которая изменилась, включая те, которые были удалены или обновлены. Например, при обновлении строки вы увидите версию удаления и версию вставки в исчерпывающих изменениях, но только версию вставки в дедуппированных измененных строках. В зависимости от варианта использования можно выбрать вариант, соответствующий вашим потребностям. Параметр по умолчанию имеет значение false, что означает исчерпывающие изменения.	No	Boolean	netChanges
Включение системных столбцов	При использовании отслеживания изменений snowflake можно использовать параметр systemColumns, чтобы контролировать, включены ли или исключены столбцы потока метаданных, предоставляемые Snowflake, в выходные данные отслеживания изменений. По умолчанию systemColumns имеет значение true, что означает, что столбцы потока метаданных включены. Если вы хотите исключить их, можно задать значение false systemColumns.	No	Boolean	systemColumns
Начало чтения с начала	Установка этого параметра с добавочным извлечением и отслеживанием изменений приведет к тому, что ADF будет считывать все строки при первом выполнении конвейера с включенным добавочным извлечением.	No	Boolean	skipInitialLoad

Примеры сценариев для источника 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 теперь поддерживает встроенную функцию в Snowflake, которая называется отслеживанием изменений, которая включает отслеживание изменений в виде журналов. Эта функция snowflake позволяет отслеживать изменения данных с течением времени, что делает его полезным для добавочной загрузки и аудита данных. Чтобы использовать эту функцию, при включении отслеживания измененных данных и выборе Отслеживание изменений Snowflake мы создадим объект Stream для исходной таблицы, которая позволяет отслеживать изменения в исходной таблице snowflake. Затем мы используем предложение CHANGES в нашем запросе для получения только новых или обновленных данных из исходной таблицы. Кроме того, рекомендуется запланировать конвейер так, чтобы изменения потреблялись в течение интервала времени хранения данных, установленного для исходной таблицы Snowflake, иначе пользователь может наблюдать несогласованное поведение в захваченных изменениях.

Преобразование приемника

В таблице, приведенной ниже, указаны свойства, поддерживаемые приемником Snowflake. Эти свойства можно изменить на вкладке "Параметры ". При использовании встроенного набора данных вы увидите дополнительные параметры, которые совпадают с свойствами, описанными в разделе свойств набора данных . Соединитель использует внутреннюю передачу данных Snowflake.

Name	Description	Required	Допустимые значения	Свойство скрипта для потока данных
Метод update	Определение операций, разрешенных в назначении Snowflake. Для обновления, объединенного обновления и вставки или удаления строк требуется преобразование Alter row для пометки строк для этих действий.	Yes	true или false	deletable insertable updateable upsertable
Ключевые столбцы	Для выполнения обновления (update), обновления или вставки (upsert) или удаления (delete) должен быть установлен ключевой столбец (или столбцы), позволяющий определить строки для изменения.	No	Array	keys
Действие таблицы	Определяет, следует ли повторно создавать или удалять все строки в целевой таблице перед записью. - Нет: Никакие действия с таблицей выполнены не будут. - Повторное создание: таблица будет удалена и создана заново. Это действие необходимо, если новая таблица создается динамически. - Усечение: все строки из целевой таблицы будут удалены.	No	true или false	recreate truncate

Примеры сценариев для приемника 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

Оптимизация pushdown запроса

При задании уровня ведения журнала конвейера значение None мы исключим передачу промежуточных метрик преобразования, предотвращая потенциальные помехи оптимизации Spark и обеспечивая оптимизацию pushdown запросов, предоставляемую Snowflake. Эта оптимизация pushdown позволяет значительно повысить производительность больших таблиц Snowflake с обширными наборами данных.

Note

Мы не поддерживаем временные таблицы в Snowflake, так как они являются локальными для сеанса или пользователя, который создает их, что делает их недоступными для других сеансов и подвержены перезаписи как обычные таблицы Snowflake. Хотя Snowflake предлагает временные таблицы в качестве альтернативы, которые доступны глобально, они требуют ручного удаления, что противоречит нашей основной цели использования временных таблиц, что позволяет избежать любых операций удаления в исходной схеме.

Сопоставление типов данных для Snowflake версии 2

При копировании данных из Snowflake используются следующие сопоставления типов данных Snowflake с промежуточными типами данных, применяемыми внутри службы. Чтобы узнать, как действие копирования сопоставляет исходную схему и типы данных с приемником, см. раздел Сопоставление схем и типов данных.

Тип данных Snowflake	Промежуточный служебный тип данных
NUMBER (p,0)	Decimal
NUMBER (p,s, где s>0)	Decimal
FLOAT	Double
VARCHAR	String
CHAR	String
BINARY	Byte[]
BOOLEAN	Boolean
DATE	DateTime
TIME	TimeSpan
TIMESTAMP_LTZ	DateTimeOffset
TIMESTAMP_NTZ	DateTimeOffset
TIMESTAMP_TZ	DateTimeOffset
VARIANT	String
OBJECT	String
ARRAY	String

Свойства действия поиска

Дополнительные сведения о свойствах см. в действии поиска.

Жизненный цикл и обновление соединителя Snowflake

В следующей таблице показаны этап выпуска и журналы изменений для различных версий соединителя Snowflake:

Version	Этап выпуска	Журнал изменений
Snowflake V1	Removed	Неприменимо.
Snowflake V2 (версия 1.0)	GA версия доступна	• Добавьте поддержку проверки подлинности пары ключей. • Добавьте поддержку storageIntegration в операции копирования. accountIdentifier, warehouse, database, schema и role свойства используются для установления соединения вместо свойства connectionstring. • Добавьте поддержку типа данных Decimal в операции Lookup. Тип NUMBER, определенный в Snowflake, будет отображаться как строка в действии подстановки. Если вы хотите преобразовать его в числовой тип в V2, можно использовать параметр pipeline с функцией int или функцией float. Например, int(activity('lookup').output.firstRow.VALUE), float(activity('lookup').output.firstRow.VALUE). • тип данных метки времени в Snowflake трактуется как тип данных DateTimeOffset в действиях "Поиск" и "Скрипт". Если после обновления до версии 2 вам все еще нужно использовать значение DateTime в качестве параметра в вашем конвейере, вы можете преобразовать тип DateTimeOffset в тип DateTime с помощью функции formatDateTime (рекомендуется) или функции concat. Например: formatDateTime(activity('lookup').output.firstRow.DATETIMETYPE), concat(substring(activity('lookup').output.firstRow.DATETIMETYPE, 0, 19), 'Z') • ЧИСЛО (p,0) считывается как десятичный тип данных. • TIMESTAMP_LTZ, TIMESTAMP_NTZ и TIMESTAMP_TZ считываются как тип данных DateTimeOffset. • Параметры скрипта не поддерживаются в действии скрипта. В качестве альтернативы используйте динамические выражения для параметров скрипта. Дополнительные сведения см. в статье "Выражения и функции" в Фабрике данных Azure и Azure Synapse Analytics. • Выполнение нескольких инструкций SQL в действии скрипта не поддерживается.
Snowflake V2 (версия 1.1)	GA версия доступна	• Добавьте поддержку параметров скрипта. • Добавлена поддержка выполнения нескольких инструкций в действии скрипта. • Добавление свойства treatDecimalAsString в активности Lookup и Script. • Добавьте дополнительное свойство подключения UseUtcTimestamps.

Обновление соединителя Snowflake с версии 1 до версии 2

Чтобы обновить соединитель Snowflake с версии 1 до версии 2, можно выполнить параллельное обновление или обновление на месте.

Параллельное обновление

Чтобы выполнить параллельное обновление, выполните следующие действия.

1. Создайте связанную службу Snowflake и настройте ее, ссылаясь на свойства связанной службы версии 2.
2. Создайте набор данных на основе созданной связанной службы Snowflake.
3. Замените новую связанную службу и набор данных существующими в конвейерах, предназначенными для объектов версии 1.

Обновление на месте

Чтобы выполнить обновление на месте, необходимо изменить существующую полезные данные связанной службы и обновить набор данных, чтобы использовать новую связанную службу.

1. Обновите тип от Snowflake до SnowflakeV2.

2. Измените полезные данные связанной службы с его формата версии 1 на V2. Вы можете заполнить каждое поле из пользовательского интерфейса после изменения указанного выше типа или обновить полезные данные непосредственно через редактор JSON. Сведения о поддерживаемых свойствах подключения см. в разделе свойств связанной службы в этой статье. В следующих примерах показаны различия полезных данных для связанных служб V1 и V2 Snowflake:

   Полезные данные JSON связанной службы Snowflake версии 1:

     {
        "name": "Snowflake1",
        "type": "Microsoft.DataFactory/factories/linkedservices",
        "properties": {
            "annotations": [],
            "type": "Snowflake",
            "typeProperties": {
                "authenticationType": "Basic",
                "connectionString": "jdbc:snowflake://<fake_account>.snowflakecomputing.com/?user=FAKE_USER&db=FAKE_DB&warehouse=FAKE_DW&schema=PUBLIC",
                "encryptedCredential": "<your_encrypted_credential_value>"
            },
            "connectVia": {
                "referenceName": "AzureIntegrationRuntime",
                "type": "IntegrationRuntimeReference"
            }
        }
    }
   

   JSON полезная нагрузка подключенной службы Snowflake V2:

    {
        "name": "Snowflake2",
        "type": "Microsoft.DataFactory/factories/linkedservices",
        "properties": {
            "parameters": {
                "schema": {
                    "type": "string",
                    "defaultValue": "PUBLIC"
                }
            },
            "annotations": [],
            "type": "SnowflakeV2",
            "typeProperties": {
                "authenticationType": "Basic",
                "accountIdentifier": "<FAKE_Account>",
                "user": "FAKE_USER",
                "database": "FAKE_DB",
                "warehouse": "FAKE_DW",
                "encryptedCredential": "<placeholder>"
            },
            "connectVia": {
                "referenceName": "AutoResolveIntegrationRuntime",
                "type": "IntegrationRuntimeReference"
            }
        }
    }
   

3. Обновите набор данных, чтобы использовать новую связанную службу. Можно создать новый набор данных на основе только что созданной связанной службы или обновить свойство типа существующего набора данных из SnowflakeTable на SnowflakeV2Table.

Note

При переходе связанных служб раздел параметров шаблона переопределения может отображать только свойства базы данных. Это можно устранить, изменив параметры вручную. После этого раздел Переопределить параметры шаблона отобразит строки подключения.

Обновление соединителя Snowflake версии 2 с версии 1.0 до версии 1.1

На странице "Изменить связанную службу " выберите 1.1 для версии. Дополнительные сведения см. в разделе "Свойства связанной службы".

Связанный контент

Для получения списка хранилищ данных, поддерживаемых в качестве источников и приемников операцией копирования, см. поддерживаемые хранилища данных и форматы.