Копирование нескольких таблиц в пакетном режиме с помощью Фабрики данных Azure и PowerShell

Статья
08/10/2023

Область применения:Фабрика данных Azure Azure Synapse Analytics

Совет

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

В этом руководстве показано, как скопировать несколько таблиц из Базы данных SQL Azure в Azure Synapse Analytics. Этот подход можно применить и в других сценариях. Например, копирование таблиц из SQL Server или Oracle в Базу данных SQL Azure, хранилище данных или большой двоичный объект Azure, копирование различных путей из большого двоичного объекта в таблицы Базы данных SQL Azure.

В целом это руководство включает следующие шаги:

Создали фабрику данных.
Создание экземпляров Базы данных SQL Azure, Azure Synapse Analytics и связанных служб хранилища Azure.
Создание Базы данных SQL Azure и наборов данных Azure Synapse Analytics.
Создание конвейера для поиска таблиц, которые нужно скопировать, и конвейера для выполнения операции копирования.
Запуск конвейера.
Мониторинг конвейера и выполнения действий.

В этом руководстве используется Azure PowerShell. Сведения об использовании других средств или пакетов SDK для создания фабрики данных см. в этом кратком руководстве.

Комплексный рабочий процесс

В этом сценарии есть несколько таблиц базы данных SQL Azure, которые необходимо скопировать в Azure Synapse Analytics. Вот логическая последовательность действий рабочего процесса в конвейерах:

Workflow

Первый конвейер ищет список таблиц, который необходимо скопировать в хранилище данных-приемник. В качестве альтернативы можно создать таблицу метаданных, в которой перечислены все таблицы, которые нужно скопировать в хранилище данных-приемник. Затем конвейер активирует другой конвейер, который выполняет итерацию по каждой таблице базы данных и выполняет операцию копирования данных.
Второй конвейер выполняет фактическое копирование. Он принимает список таблиц в качестве параметра. Чтобы добиться лучшей производительности, для каждой таблицы в списке скопируйте определенную таблицу из Базы данных SQL Azure в соответствующую таблицу Azure Synapse Analytics, используя промежуточное копирование с помощью хранилища BLOB-объектов и PolyBase. В этом примере первый конвейер передает список таблиц в качестве значения параметра.

Если у вас нет подписки Azure, создайте бесплатную учетную запись, прежде чем приступить к работе.

Предварительные требования

Примечание.

Мы рекомендуем использовать модуль Azure Az PowerShell для взаимодействия с Azure. Чтобы начать работу, см. статью Установка Azure PowerShell. Дополнительные сведения см. в статье Перенос Azure PowerShell с AzureRM на Az.

Azure PowerShell. Следуйте инструкциям по установке и настройке Azure PowerShell.
Учетная запись хранения Azure. Учетная запись хранения Azure используется в качестве промежуточного хранилища больших двоичных объектов в операции массового копирования.
База данных SQL Azure. Эта база данных содержит исходные данные.
Azure Synapse Analytics. Это хранилище данных содержит данные, копируемые из базы данных SQL.

Подготовка Базы данных SQL Azure и Azure Synapse Analytics

Подготовка исходной базы данных SQL Azure:

Создайте базу данных в службе "База данных SQL", используя пример данных Adventure Works LT, представленный в статье Создание базы данных в службе "База данных SQL Azure". В этом учебнике все таблицы из этого примера базы данных копируются в Azure Synapse Analytics.

Подготовка приемника в Azure Synapse Analytics.

Если у вас нет рабочей области Azure Synapse Analytics, создайте ее по инструкциям из статьи Начало работы с Azure Synapse Analytics.
Создание соответствующих схем таблиц в Azure Synapse Analytics. Фабрика данных Azure используется для миграции и копирования данных на более поздних этапах.

Доступ служб Azure к серверу SQL Server

В Базе данных SQL и Azure Synapse Analytics предоставьте службам Azure доступ к серверу SQL Server. Убедитесь, что параметр Разрешить доступ к службам Azureвключен для вашего сервера. Этот параметр позволяет службе Фабрики данных читать данные из Базы данных SQL Azure и записывать их в Azure Synapse Analytics. Чтобы проверить и при необходимости включить этот параметр, сделайте следующее.

Щелкните Все службы слева и выберите Серверы SQL.
Выберите сервер и щелкните Брандмауэр в разделе Параметры.
На странице Параметры брандмауэра щелкните ВКЛ для параметра Разрешить доступ к службам Azure.

Создание фабрики данных

Запустите PowerShell. Не закрывайте Azure PowerShell, пока выполняются описанные в учебнике инструкции. Если закрыть и снова открыть это окно, то придется вновь выполнять эти команды.

Выполните следующую команду и введите имя пользователя и пароль, которые используются для входа на портал Azure.
```
Connect-AzAccount
```
Чтобы просмотреть все подписки для этой учетной записи, выполните следующую команду:
```
Get-AzSubscription
```
Выполните следующую команду, чтобы выбрать подписку, с которой вы собираетесь работать. Замените значение SubscriptionId на идентификатор подписки Azure:
```
Select-AzSubscription -SubscriptionId "<SubscriptionId>"
```
Выполните командлет Set-AzDataFactoryV2, чтобы создать фабрику данных. Перед выполнением команды замените заполнители собственными значениями.
```
$resourceGroupName = "<your resource group to create the factory>"
$dataFactoryName = "<specify the name of data factory to create. It must be globally unique.>"
Set-AzDataFactoryV2 -ResourceGroupName $resourceGroupName -Location "East US" -Name $dataFactoryName
```
Обратите внимание на следующие аспекты:
- Имя фабрики данных Azure должно быть глобально уникальным. Если появляется следующая ошибка, измените имя и повторите попытку.
```
The specified Data Factory name 'ADFv2QuickStartDataFactory' is already in use. Data Factory names must be globally unique.
```
- Чтобы создавать экземпляры фабрики данных, вы должны быть участником или администратором подписки Azure.
- Чтобы получить список регионов Azure, в которых в настоящее время доступна Фабрика данных, выберите интересующие вас регионы на следующей странице, а затем разверните раздел Аналитика, чтобы найти пункт Фабрика данных: Доступность продуктов по регионам. Хранилища данных (служба хранилища Azure, база данных SQL Azure и т. д.) и вычисления (HDInsight и т. д.), используемые фабрикой данных, могут располагаться в других регионах.

Создание связанных служб

В этом руководстве создаются три связанные службы для источника, приемника и промежуточного большого двоичного объекта соответственно, включая подключения к хранилищам данных:

Создание исходной связанной службы Базы данных SQL Azure

Создайте файл JSON с именем AzureSqlDatabaseLinkedService.json в папке C:\ADFv2TutorialBulkCopy и добавьте в него приведенное ниже содержимое. Если папка ADFv2TutorialBulkCopy отсутствует, создайте ее.

Важно!

Перед сохранением файла замените заполнители <servername>, <databasename>, <username>@<servername> и <password> значениями для используемой Базы данных SQL Azure.
```
{
    "name": "AzureSqlDatabaseLinkedService",
    "properties": {
        "type": "AzureSqlDatabase",
        "typeProperties": {
            "connectionString": "Server=tcp:<servername>.database.windows.net,1433;Database=<databasename>;User ID=<username>@<servername>;Password=<password>;Trusted_Connection=False;Encrypt=True;Connection Timeout=30"
        }
    }
}
```
В Azure PowerShell переключитесь в папку ADFv2TutorialBulkCopy.

Выполните командлет Set-AzDataFactoryV2LinkedService, чтобы создать связанную службу: AzureSqlDatabaseLinkedService.

Set-AzDataFactoryV2LinkedService -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -Name "AzureSqlDatabaseLinkedService" -File ".\AzureSqlDatabaseLinkedService.json"

Пример выходных данных:

LinkedServiceName : AzureSqlDatabaseLinkedService
ResourceGroupName : <resourceGroupName>
DataFactoryName   : <dataFactoryName>
Properties        : Microsoft.Azure.Management.DataFactory.Models.AzureSqlDatabaseLinkedService

Создание приемника связанной службы Azure Synapse Analytics

Создайте файл JSON с именем AzureSqlDWLinkedService.json в папке C:\ADFv2TutorialBulkCopy со следующим содержимым:

Важно!

Перед сохранением файла замените заполнители <servername>, <databasename>, <username>@<servername> и <password> значениями для используемой Базы данных SQL Azure.
```
{
    "name": "AzureSqlDWLinkedService",
    "properties": {
        "type": "AzureSqlDW",
        "typeProperties": {
            "connectionString": "Server=tcp:<servername>.database.windows.net,1433;Database=<databasename>;User ID=<username>@<servername>;Password=<password>;Trusted_Connection=False;Encrypt=True;Connection Timeout=30"
        }
    }
}
```

Чтобы создать связанную службу: AzureSqlDWLinkedService, выполните командлет Set-AzDataFactoryV2LinkedService.

Set-AzDataFactoryV2LinkedService -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -Name "AzureSqlDWLinkedService" -File ".\AzureSqlDWLinkedService.json"

Пример выходных данных:

LinkedServiceName : AzureSqlDWLinkedService
ResourceGroupName : <resourceGroupName>
DataFactoryName   : <dataFactoryName>
Properties        : Microsoft.Azure.Management.DataFactory.Models.AzureSqlDWLinkedService

Создание промежуточной связанной службы хранилища Azure

В этом руководстве хранилища BLOB-объектов Azure используются в качестве области промежуточного хранения, чтобы включить PolyBase для повышения производительности копирования.

Создайте файл JSON с именем AzureStorageLinkedService.json в папке C:\ADFv2TutorialBulkCopy со следующим содержимым:

Важно!

Перед сохранением файла замените значения <accountname> и <accountkey> на имя вашей учетной записи хранения Azure и ее ключ.
```
{
    "name": "AzureStorageLinkedService",
    "properties": {
        "type": "AzureStorage",
        "typeProperties": {
            "connectionString": "DefaultEndpointsProtocol=https;AccountName=<accountName>;AccountKey=<accountKey>"
        }
    }
}
```

Чтобы создать связанную службу: Azure служба хранилища LinkedService, выполните командлет Set-AzDataFactoryV2LinkedService.

Set-AzDataFactoryV2LinkedService -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -Name "AzureStorageLinkedService" -File ".\AzureStorageLinkedService.json"

Пример выходных данных:

LinkedServiceName : AzureStorageLinkedService
ResourceGroupName : <resourceGroupName>
DataFactoryName   : <dataFactoryName>
Properties        : Microsoft.Azure.Management.DataFactory.Models.AzureStorageLinkedService

Создайте наборы данных.

В этом руководстве создаются наборы данных источника и приемника, в которых указывается место хранения данных:

Создание набора данных для исходной базы данных SQL

Создайте файл JSON с именем AzureSqlDatabaseDataset.json в папке C:\ADFv2TutorialBulkCopy со следующим содержимым. TableName является фиктивным,так как позже в действии копирования будет использован SQL-запрос для извлечения данных.
```
{
    "name": "AzureSqlDatabaseDataset",
    "properties": {
        "type": "AzureSqlTable",
        "linkedServiceName": {
            "referenceName": "AzureSqlDatabaseLinkedService",
            "type": "LinkedServiceReference"
        },
        "typeProperties": {
            "tableName": "dummy"
        }
    }
}
```

Чтобы создать набор данных: AzureSqlDatabaseDataset, выполните командлет Set-AzDataFactoryV2Dataset .

Set-AzDataFactoryV2Dataset -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -Name "AzureSqlDatabaseDataset" -File ".\AzureSqlDatabaseDataset.json"

Пример выходных данных:

DatasetName       : AzureSqlDatabaseDataset
ResourceGroupName : <resourceGroupname>
DataFactoryName   : <dataFactoryName>
Structure         :
Properties        : Microsoft.Azure.Management.DataFactory.Models.AzureSqlTableDataset

Создание набора данных для приемника Azure Synapse Analytics

Создайте файл JSON с именем AzureSqlDWDataset.json в папке C:\ADFv2TutorialBulkCopy со следующим содержимым: tableName задается в качестве параметра, позже действие копирования, которое ссылается на этот набор данных, передает фактическое значение в набор данных.

{
    "name": "AzureSqlDWDataset",
    "properties": {
        "type": "AzureSqlDWTable",
        "linkedServiceName": {
            "referenceName": "AzureSqlDWLinkedService",
            "type": "LinkedServiceReference"
        },
        "typeProperties": {
            "tableName": {
                "value": "@{dataset().DWTableName}",
                "type": "Expression"
            }
        },
        "parameters":{
            "DWTableName":{
                "type":"String"
            }
        }
    }
}

Чтобы создать набор данных: AzureSqlDWDataset, выполните командлет Set-AzDataFactoryV2Dataset .

Set-AzDataFactoryV2Dataset -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -Name "AzureSqlDWDataset" -File ".\AzureSqlDWDataset.json"

Пример выходных данных:

DatasetName       : AzureSqlDWDataset
ResourceGroupName : <resourceGroupname>
DataFactoryName   : <dataFactoryName>
Structure         :
Properties        : Microsoft.Azure.Management.DataFactory.Models.AzureSqlDwTableDataset

Создание конвейеров

В этом руководстве создается два конвейера:

Создание конвейера IterateAndCopySQLTables

Этот конвейер принимает список таблиц в качестве параметра. Для каждой таблицы в списке он копирует данные из таблицы в Базе данных SQL Azure в Azure Synapse Analytics с помощью промежуточного копирования и PolyBase.

Создайте файл JSON с именем IterateAndCopySQLTables.json в папке C:\ADFv2TutorialBulkCopy со следующим содержимым:

{
    "name": "IterateAndCopySQLTables",
    "properties": {
        "activities": [
            {
                "name": "IterateSQLTables",
                "type": "ForEach",
                "typeProperties": {
                    "isSequential": "false",
                    "items": {
                        "value": "@pipeline().parameters.tableList",
                        "type": "Expression"
                    },
                    "activities": [
                        {
                            "name": "CopyData",
                            "description": "Copy data from Azure SQL Database to Azure Synapse Analytics",
                            "type": "Copy",
                            "inputs": [
                                {
                                    "referenceName": "AzureSqlDatabaseDataset",
                                    "type": "DatasetReference"
                                }
                            ],
                            "outputs": [
                                {
                                    "referenceName": "AzureSqlDWDataset",
                                    "type": "DatasetReference",
                                    "parameters": {
                                        "DWTableName": "[@{item().TABLE_SCHEMA}].[@{item().TABLE_NAME}]"
                                    }
                                }
                            ],
                            "typeProperties": {
                                "source": {
                                    "type": "SqlSource",
                                    "sqlReaderQuery": "SELECT * FROM [@{item().TABLE_SCHEMA}].[@{item().TABLE_NAME}]"
                                },
                                "sink": {
                                    "type": "SqlDWSink",
                                    "preCopyScript": "TRUNCATE TABLE [@{item().TABLE_SCHEMA}].[@{item().TABLE_NAME}]",
                                    "allowPolyBase": true
                                },
                                "enableStaging": true,
                                "stagingSettings": {
                                    "linkedServiceName": {
                                        "referenceName": "AzureStorageLinkedService",
                                        "type": "LinkedServiceReference"
                                    }
                                }
                            }
                        }
                    ]
                }
            }
        ],
        "parameters": {
            "tableList": {
                "type": "Object"
            }
        }
    }
}

Чтобы создать конвейер: IterateAndCopySQLTables, выполните командлет Set-AzDataFactoryV2Pipeline .

Set-AzDataFactoryV2Pipeline -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -Name "IterateAndCopySQLTables" -File ".\IterateAndCopySQLTables.json"

Пример выходных данных:

PipelineName      : IterateAndCopySQLTables
ResourceGroupName : <resourceGroupName>
DataFactoryName   : <dataFactoryName>
Activities        : {IterateSQLTables}
Parameters        : {[tableList, Microsoft.Azure.Management.DataFactory.Models.ParameterSpecification]}

Создание конвейера GetTableListAndTriggerCopyData

Этот конвейер выполняет два действия:

Ищет системную таблицу базы данных SQL Azure, чтобы получить список таблиц для копирования.
Активирует конвейер IterateAndCopySQLTables для выполнения копирования данных.

Создайте файл JSON с именем GetTableListAndTriggerCopyData.json в папке C:\ADFv2TutorialBulkCopy со следующим содержимым:

{
    "name":"GetTableListAndTriggerCopyData",
    "properties":{
        "activities":[
            { 
                "name": "LookupTableList",
                "description": "Retrieve the table list from Azure SQL dataabse",
                "type": "Lookup",
                "typeProperties": {
                    "source": {
                        "type": "SqlSource",
                        "sqlReaderQuery": "SELECT TABLE_SCHEMA, TABLE_NAME FROM information_schema.TABLES WHERE TABLE_TYPE = 'BASE TABLE' and TABLE_SCHEMA = 'SalesLT' and TABLE_NAME <> 'ProductModel'"
                    },
                    "dataset": {
                        "referenceName": "AzureSqlDatabaseDataset",
                        "type": "DatasetReference"
                    },
                    "firstRowOnly": false
                }
            },
            {
                "name": "TriggerCopy",
                "type": "ExecutePipeline",
                "typeProperties": {
                    "parameters": {
                        "tableList": {
                            "value": "@activity('LookupTableList').output.value",
                            "type": "Expression"
                        }
                    },
                    "pipeline": {
                        "referenceName": "IterateAndCopySQLTables",
                        "type": "PipelineReference"
                    },
                    "waitOnCompletion": true
                },
                "dependsOn": [
                    {
                        "activity": "LookupTableList",
                        "dependencyConditions": [
                            "Succeeded"
                        ]
                    }
                ]
            }
        ]
    }
}

Чтобы создать конвейер: GetTableListAndTriggerCopyData, выполните командлет Set-AzDataFactoryV2Pipeline .

Set-AzDataFactoryV2Pipeline -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -Name "GetTableListAndTriggerCopyData" -File ".\GetTableListAndTriggerCopyData.json"

Пример выходных данных:

PipelineName      : GetTableListAndTriggerCopyData
ResourceGroupName : <resourceGroupName>
DataFactoryName   : <dataFactoryName>
Activities        : {LookupTableList, TriggerCopy}
Parameters        :

Запуск и мониторинг выполнения конвейера

Запустите выполнение главного конвейера GetTableListAndTriggerCopyData и запишите идентификатор выполнения конвейера для будущего мониторинга. В рамках конвейера запускается выполнение конвейера IterateAndCopySQLTables, как указано в действии ExecutePipeline.
```
$runId = Invoke-AzDataFactoryV2Pipeline -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -PipelineName 'GetTableListAndTriggerCopyData'
```

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

while ($True) {
    $run = Get-AzDataFactoryV2PipelineRun -ResourceGroupName $resourceGroupName -DataFactoryName $DataFactoryName -PipelineRunId $runId

    if ($run) {
        if ($run.Status -ne 'InProgress') {
            Write-Host "Pipeline run finished. The status is: " $run.Status -ForegroundColor "Yellow"
            Write-Host "Pipeline run details:" -ForegroundColor "Yellow"
            $run
            break
        }
        Write-Host  "Pipeline is running...status: InProgress" -ForegroundColor "Yellow"
    }

    Start-Sleep -Seconds 15
}

$result = Get-AzDataFactoryV2ActivityRun -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -PipelineRunId $runId -RunStartedAfter (Get-Date).AddMinutes(-30) -RunStartedBefore (Get-Date).AddMinutes(30)
Write-Host "Activity run details:" -ForegroundColor "Yellow"
$result

Вот результат примера выполнения:

Pipeline run details:
ResourceGroupName : <resourceGroupName>
DataFactoryName   : <dataFactoryName>
RunId             : 0000000000-00000-0000-0000-000000000000
PipelineName      : GetTableListAndTriggerCopyData
LastUpdated       : 9/18/2017 4:08:15 PM
Parameters        : {}
RunStart          : 9/18/2017 4:06:44 PM
RunEnd            : 9/18/2017 4:08:15 PM
DurationInMs      : 90637
Status            : Succeeded
Message           : 

Activity run details:
ResourceGroupName : <resourceGroupName>
DataFactoryName   : <dataFactoryName>
ActivityName      : LookupTableList
PipelineRunId     : 0000000000-00000-0000-0000-000000000000
PipelineName      : GetTableListAndTriggerCopyData
Input             : {source, dataset, firstRowOnly}
Output            : {count, value, effectiveIntegrationRuntime}
LinkedServiceName : 
ActivityRunStart  : 9/18/2017 4:06:46 PM
ActivityRunEnd    : 9/18/2017 4:07:09 PM
DurationInMs      : 22995
Status            : Succeeded
Error             : {errorCode, message, failureType, target}

ResourceGroupName : <resourceGroupName>
DataFactoryName   : <dataFactoryName>
ActivityName      : TriggerCopy
PipelineRunId     : 0000000000-00000-0000-0000-000000000000
PipelineName      : GetTableListAndTriggerCopyData
Input             : {pipeline, parameters, waitOnCompletion}
Output            : {pipelineRunId}
LinkedServiceName : 
ActivityRunStart  : 9/18/2017 4:07:11 PM
ActivityRunEnd    : 9/18/2017 4:08:14 PM
DurationInMs      : 62581
Status            : Succeeded
Error             : {errorCode, message, failureType, target}

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

Write-Host "Pipeline 'IterateAndCopySQLTables' run result:" -ForegroundColor "Yellow"
($result | Where-Object {$_.ActivityName -eq "TriggerCopy"}).Output.ToString()

Вот результат примера выполнения:

{
    "pipelineRunId": "7514d165-14bf-41fb-b5fb-789bea6c9e58"
}

$result2 = Get-AzDataFactoryV2ActivityRun -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -PipelineRunId <copy above run ID> -RunStartedAfter (Get-Date).AddMinutes(-30) -RunStartedBefore (Get-Date).AddMinutes(30)
$result2

Подключитесь к приемнику Azure Synapse Analytics и подтвердите, что данные из базы данных SQL Azure скопированы надлежащим образом.

В этом руководстве вы выполнили следующие шаги:

Создали фабрику данных.
Создание экземпляров Базы данных SQL Azure, Azure Synapse Analytics и связанных служб хранилища Azure.
Создание Базы данных SQL Azure и наборов данных Azure Synapse Analytics.
Создание конвейера для поиска таблиц, которые нужно скопировать, и конвейера для выполнения операции копирования.
Запуск конвейера.
Мониторинг конвейера и выполнения действий.

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

Пошаговая загрузка данных из базы данных SQL Azure в хранилище BLOB-объектов Azure

Копирование нескольких таблиц в пакетном режиме с помощью Фабрики данных Azure и PowerShell

Комплексный рабочий процесс

Предварительные требования

Подготовка Базы данных SQL Azure и Azure Synapse Analytics

Доступ служб Azure к серверу SQL Server

Создание фабрики данных

Создание связанных служб

Создание исходной связанной службы Базы данных SQL Azure

Создание приемника связанной службы Azure Synapse Analytics

Создание промежуточной связанной службы хранилища Azure

Создайте наборы данных.

Создание набора данных для исходной базы данных SQL

Создание набора данных для приемника Azure Synapse Analytics

Создание конвейеров

Создание конвейера IterateAndCopySQLTables

Создание конвейера GetTableListAndTriggerCopyData

Запуск и мониторинг выполнения конвейера

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

Дополнительные ресурсы