Массовое копирование нескольких таблиц с помощью Azure Data Factory с помощью PowerShell

ПРИМЕНИМО К: Azure Data Factory Azure Synapse Analytics

Совет

Data Factory в Microsoft Fabric — это следующее поколение Azure Data Factory с более простой архитектурой, встроенным ИИ и новыми функциями. Если вы не знакомы с интеграцией данных, начните с Fabric Data Factory. Существующие рабочие нагрузки ADF могут обновляться до Fabric для доступа к новым возможностям в области обработки и анализа данных, аналитики в режиме реального времени и отчетов.

В этом руководстве демонстрируется копирование ряда таблиц из Azure SQL Database в Azure Synapse Analytics. Тот же подход можно использовать и в других сценариях копирования. Например, копирование таблиц из SQL Server/Oracle в Azure SQL Database/Data Warehouse/Azure Blob, копирование различных путей из Azure Blob в таблицы Azure SQL Database.

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

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

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

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

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

Рабочий процесс

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

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

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

Примечание.

Мы рекомендуем использовать модуль Az PowerShell Azure для взаимодействия с Azure. Сведения о начале работы см. в разделе Install Azure PowerShell. Сведения о миграции в модуль Az PowerShell см. в статье Migrate Azure PowerShell из AzureRM в Az.

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

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

Подготовьте исходную базу данных Azure SQL:

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

Подготовка синка Azure Synapse Analytics:

  1. Если у вас нет рабочей области Azure Synapse Analytics, ознакомьтесь со статьей Get started with Azure Synapse Analytics для получения инструкций по её созданию.

  2. Создайте соответствующие схемы таблиц в Azure Synapse Analytics. Вы используете Azure Data Factory для переноса и копирования данных на более поздних шагах.

Службы Azure для доступа к SQL Server

Для базы данных SQL и Azure Synapse Analytics разрешите Azure службам доступ к SQL Server. Убедитесь, что параметр Разрешить доступ к службам Azure включен ВКЛЮЧЕНО для вашего сервера. Этот параметр позволяет службе фабрики данных считывать данные из Azure SQL Database и записывать данные в Azure Synapse Analytics. Чтобы проверить и при необходимости включить этот параметр, сделайте следующее.

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

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

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

    Выполните следующую команду и введите имя пользователя и пароль, которые вы используете для входа на портал Azure:

    Connect-AzAccount

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

    Get-AzSubscription

    Выполните следующую команду, чтобы выбрать подписку, с которой вы собираетесь работать. Замените SubscriptionId идентификатором подписки Azure:

    Select-AzSubscription -SubscriptionId "<SubscriptionId>"

  2. Выполните командлет 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.

    • Чтобы создать экземпляры Data Factory, необходимо быть Соавтором или Администратором подписки Azure.

    • В списке регионов Azure, в которых в настоящее время доступен Data Factory, выберите интересующие вас регионы на следующей странице, а затем разверните раздел Analytics, чтобы найти Data Factory: Продукты, доступные по регионам. Хранилища данных (Azure Storage, Azure SQL Database и т. д.) и вычислительные ресурсы (HDInsight и т. д.), используемые фабрикой данных, могут находиться в других регионах.

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

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

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

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

    Внимание

    Замените <servername>, <databasename><username>@<servername> и <password> со значениями Azure SQL Database перед сохранением файла.

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

  2. В Azure PowerShell перейдите в папку ADFv2TutorialBulkCopy.

  3. Выполните командлет 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

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

    Внимание

    Замените <servername>, <databasename><username>@<servername> и <password> со значениями Azure SQL Database перед сохранением файла.

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

  2. Чтобы создать связанную службу: 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 Storage

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

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

    Внимание

    Замените <accountName> и <accountKey> именем и ключом учетной записи хранения Azure перед сохранением файла.

    {
    "name": "AzureStorageLinkedService",
    "properties": {
        "type": "AzureStorage",
        "typeProperties": {
            "connectionString": "DefaultEndpointsProtocol=https;AccountName=<accountName>;AccountKey=<accountKey>"
        }
    }
}

  2. Чтобы создать связанную службу AzureStorageLinkedService, выполните командлет 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

  1. Создайте файл JSON с именем AzureSqlDatabaseDataset.json в папке C:\ADFv2TutorialBulkCopy со следующим содержимым. TableName является фиктивным,так как позже в действии копирования будет использован SQL-запрос для извлечения данных.

    {
    "name": "AzureSqlDatabaseDataset",
    "properties": {
        "type": "AzureSqlTable",
        "linkedServiceName": {
            "referenceName": "AzureSqlDatabaseLinkedService",
            "type": "LinkedServiceReference"
        },
        "typeProperties": {
            "tableName": "dummy"
        }
    }
}

  2. Чтобы создать набор данных AzureSqlDatabaseDataset, запустите cmdlet 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

  1. Создайте файл 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"
            }
        }
    }
}

  2. Чтобы создать набор данных: 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"

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

  1. Создайте файл 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"
            }
        }
    }
}

  2. Чтобы создать конвейер: 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

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

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

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

    {
    "name":"GetTableListAndTriggerCopyData",
    "properties":{
        "activities":[
            { 
                "name": "LookupTableList",
                "description": "Retrieve the table list from Azure SQL database",
                "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"
                        ]
                    }
                ]
            }
        ]
    }
}

  2. Для того чтобы создать конвейер GetTableListAndTriggerCopyData, выполните командлет Set-AzDataFactoryV2Pipeline.

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

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

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

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

  1. Запустите выполнение главного конвейера GetTableListAndTriggerCopyData и запишите идентификатор выполнения конвейера для будущего мониторинга. Впоследствии запускается выполнение конвейера "IterateAndCopySQLTables", как предусмотрено в действии ExecutePipeline.

    $runId = Invoke-AzDataFactoryV2Pipeline -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -PipelineName 'GetTableListAndTriggerCopyData'

  2. Выполните следующий скрипт, чтобы постоянно проверять состояние выполнения конвейера 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}

  3. Вы можете получить идентификатор выполнения конвейера 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

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

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

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

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

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

Инкрементное копирование данных

  • Last updated on