Создание конвейеров крупномасштабного копирования данных с управляемым метаданными подходом в средстве копирования данных
Область применения:
Фабрика данных Azure Azure Synapse Analytics
Совет
Попробуйте использовать фабрику данных в Microsoft Fabric, решение для аналитики с одним интерфейсом для предприятий. Microsoft Fabric охватывает все, от перемещения данных до обработки и анализа данных в режиме реального времени, бизнес-аналитики и отчетности. Узнайте, как бесплатно запустить новую пробную версию !
Если вам нужно копировать огромные объемы объектов (например, тысячи таблиц) или загружать данные из большого спектра источников, следует ввести список имен объектов с требуемыми режимами копирования в таблицу управления, а затем использовать параметризованные конвейеры для чтения данных из этой таблицы и их применения к соответствующим заданиям. Таким образом можно оперировать списком объектов (например, дополнять его или сокращать), чтобы упростить копирование и просто обновлять имена объектов в таблице управления, а не повторно развертывать конвейеры. Более того, вы сможете легко проверить, какие объекты скопированы какими конвейерами и триггерами с определенными режимами копирования.
Средство копирования данных в ADF упрощает процесс создания таких управляемых метаданными конвейеров копирования данных. После того как вы выполните интуитивно понятные инструкции по работе с мастером, средство сможет создавать параметризованные конвейеры и сценарии SQL, чтобы соответствующим образом создать внешние таблицы управления. После запуска созданных сценариев для создания таблицы управления в базе данных SQL конвейеры будут считывать метаданные из таблицы управления и автоматически применять их к заданиям копирования.
Создание управляемых метаданными заданий копирования из средства копирования данных
В средстве копирования данных выберите Metadata-driven copy task (Управляемое метаданными задание копирования).
Необходимо указать подключение и имя таблицы управления, чтобы созданный конвейер считывал метаданные из нее.
Введите подключение к базе данных-источнику. Можно также использовать параметризованную связанную службу.
Выберите имя таблицы для копирования.
Примечание.
Если выбрать хранилище табличных данных, то на следующей странице вы сможете дополнительно выбрать полную загрузку или разностную загрузку. Если выбрать хранилище, то на следующей странице можно будет выбрать только полную загрузку. Добавочная загрузка новых файлов только из хранилища в настоящее время не поддерживается.
Выберите режим загрузки.
Совет
Если необходимо выполнить полное копирование всех таблиц, выберите Full load all tables (Полная загрузка всех таблиц). Если нужно выполнить добавочное копирование, можно выбрать параметр configure for each table individually (Настроить для каждой таблицы по отдельности) и выбрать Delta load (Разностная нагрузка), а также имя и начальное значение столбца предела для каждой таблицы.
Выберите целевое хранилище данных.
На странице Settings (Параметры) можно выбрать максимальное число действий копирования, которые будут одновременно копировать данные из исходного хранилища, с помощью параметра Number of concurrent copy tasks (Количество параллельных задач копирования). Значение по умолчанию — 20.
После развертывания конвейера можно скопировать или скачать сценарии SQL из пользовательского интерфейса для создания управляющей таблицы и процедуры хранилища.
Вы увидите два сценария SQL.
- Первый сценарий SQL используется для создания двух управляющих таблиц. В основной таблице управления хранятся список таблиц, путь к файлам или режимы копирования. В таблице управления подключениями хранится значение подключения для хранилища данных, если используется параметризованная связанная служба.
- Второй сценарий SQL используется для создания процедуры хранилища. Она используется для обновления значения предела в основной таблице управления после каждого выполнения заданий добавочного копирования.
Откройте SSMS, чтобы подключиться к серверу таблицы управления, и выполните два сценария SQL для создания управляющих таблиц и процедуры хранилища.
Выполните запрос к основной таблице управления и таблице управления подключениями, чтобы просмотреть содержащиеся в них метаданные.
Основная таблица управления
таблица управления Подключение ion
Вернитесь на портал ADF, чтобы просмотреть и отладить конвейеры. Вы увидите созданную папку с именем MetadataDrivenCopyTask_#########. Щелкните имя конвейера MetadataDrivenCopyTask###_TopLevel и выберите debug run (Выполнить отладку).
Вам потребуется ввести следующие параметры.
Имя параметра Description MaxNumberOfConcurrentTasks Вы всегда можете изменить максимальное число одновременных операций копирования перед запуском конвейера. Значение по умолчанию указывается в средстве копирования данных. MainControlTableName Вы всегда можете изменить имя основной таблицы управления, и конвейер будет получать метаданные из этой таблицы перед запуском. ConnectionControlTableName Вы всегда можете изменить имя таблицы управления подключениями (необязательно), и перед запуском конвейер будет получать из нее метаданные, связанные с подключением к хранилищам данных. MaxNumberOfObjectsReturnedFromLookupActivity Чтобы избежать достижения предельного количества операций поиска в выходных данных, можно определить максимальное число объектов, возвращаемых действием поиска. В большинстве случаев значение по умолчанию не требуется изменять. windowStart Если вы указали динамическое значение (например, "yyyy/mm/dd") в пути к папке, то этот параметр используется для передачи текущего времени активации в конвейер, чтобы заполнить динамический путь к папке. Когда конвейер активируется триггером по расписанию или периодическим триггером, пользователям не нужно вводить значение этого параметра. Пример значения: 2021-01-25T01:49:28Z Включите этот триггер, чтобы обеспечить работу конвейеров.
Обновление таблицы управления с помощью средства копирования данных
Вы всегда можете напрямую обновить таблицу управления, добавив или удалив объект, который необходимо скопировать, или изменив режим копирования для каждой таблицы. Мы также создали интерфейс пользователя в средстве копирования данных, чтобы упростить изменение таблицы управления.
Щелкните правой кнопкой мыши конвейер верхнего уровня MetadataDrivenCopyTask_xxx_TopLevel, а затем выберите Edit control table (Изменить таблицу управления).
Выберите строки из таблицы управления для изменения.
Выполните инструкции в средстве копирования данных и получите новый сценарий SQL. Запустите этот сценарий SQL, чтобы обновить таблицу управления.
Примечание.
Конвейер не будет повторно развернут. Новый сценарий SQL обновляет только таблицу управления.
Таблицы управления
Основная таблица управления
Каждая строка в таблице управления содержит метаданные одного копируемого объекта (например, таблицы).
| Имя столбца | Описание |
|---|---|
| Идентификатор | Уникальный идентификатор копируемого объекта. |
| SourceObjectSettings | Метаданные исходного набора данных. Это может быть имя схемы, имя таблицы и т. д. Ниже приведен пример. |
| SourceConnectionSettingsName | Имя параметра исходного подключения в таблице управления подключениями. Этот параметр необязателен. |
| CopySourceSettings | Метаданные свойства источника в действии копирования. Это может быть запрос, секции и т. д. Ниже приведен пример. |
| SinkObjectSettings | Метаданные целевого набора данных. Это может быть имя файла, путь к папке, имя таблицы и т. д. Ниже приведен пример. Если указан динамический путь к папке, то значение переменной не будет записываться в таблицу управления. |
| SinkConnectionSettingsName | Имя параметра целевого подключения в таблице управления подключениями. Этот параметр необязателен. |
| CopySinkSettings | Метаданные свойства приемника в действии копирования. Это может быть preCopyScript, tableOption и т. д. Ниже приведен пример. |
| CopyActivitySettings | Метаданные свойства транслятора в действии копирования. Оно используется для определения сопоставления столбцов. |
| TopLevelPipelineName | Имя конвейера верхнего уровня, который может скопировать этот объект. |
| TriggerName | Имя триггера, который может активировать конвейер для копирования этого объекта. При запуске отладки используется имя Sandbox. При выполнении вручную используется имя Manual. При запланированном выполнении используется имя связанного триггера. Может принимать несколько входных имен. |
| DataLoadingBehaviorSettings | Полная загрузка или разностная загрузка. |
| TaskId | Порядок копирования объектов после элемента TaskId в таблице управления (ORDER BY [TaskId] DESC). Если вам нужно скопировать огромное количество объектов, но количество одновременно копируемых объектов ограничено, можно изменить TaskId для каждого объекта, чтобы решить, какие объекты можно скопировать раньше. Значение по умолчанию равно 0. |
| CopyEnabled | Укажите, участвует ли элемент в процессе приема данных. Допустимые значения: 1 (участвует), 0 (не участвует). Значение по умолчанию — 1. |
Таблица управления подключениями
Каждая строка в этой таблице управления содержит один параметр подключения к хранилищу данных.
| Имя столбца | Description |
|---|---|
| Имя. | Имя параметризованного подключения в основной таблице управления. |
| ConnectionSettings | Параметры подключения. Это может быть имя базы данных, имя сервера и т. д. |
Конвейеры
Вы увидите три уровня конвейеров, созданных средством копирования данных.
MetadataDrivenCopyTask_xxx_TopLevel
Этот конвейер вычислит общее количество объектов (таблиц и т. д.), которые должны быть скопированы в этом запуске, получит количество последовательных пакетов на основе максимального числа одновременных задач копирования, а затем выполнит другой конвейер для последовательного копирования разных пакетов.
Параметры
| Имя параметра | Description |
|---|---|
| MaxNumberOfConcurrentTasks | Вы всегда можете изменить максимальное число одновременных операций копирования перед запуском конвейера. Значение по умолчанию указывается в средстве копирования данных. |
| MainControlTableName | Имя основной таблицы управления. Конвейер получит метаданные из этой таблицы перед запуском. |
| ConnectionControlTableName | Имя таблицы управления подключениями (необязательно). Перед запуском конвейер получит метаданные, связанные с подключением к хранилищам данных. |
| MaxNumberOfObjectsReturnedFromLookupActivity | Чтобы избежать достижения предельного количества операций поиска в выходных данных, можно определить максимальное число объектов, возвращаемых действием поиска. В большинстве случаев значение по умолчанию не требуется изменять. |
| windowStart | Если вы указали динамическое значение (например, "yyyy/mm/dd") в пути к папке, то этот параметр используется для передачи текущего времени активации в конвейер, чтобы заполнить динамический путь к папке. Когда конвейер активируется триггером по расписанию или периодическим триггером, пользователям не нужно вводить значение этого параметра. Пример значения: 2021-01-25T01:49:28Z |
Процедуры
| Имя действия | Тип активности | Description |
|---|---|---|
| GetSumOfObjectsToCopy | Подстановка | Вычисление общего количества объектов (таблиц и т. д.), которые должны быть скопированы при выполнении этого запуска. |
| CopyBatchesOfObjectsSequentially | ForEach | Получение количество последовательных пакетов на основе максимального количества одновременных задач копирования, а затем выполнение другого конвейера для последовательного копирования пакетов. |
| CopyObjectsInOneBtach | Выполнение конвейера | Выполнение другого конвейера, чтобы скопировать один пакет объектов. Объекты, относящиеся к этому пакету, будут копироваться параллельно. |
MetadataDrivenCopyTask_xxx_ MiddleLevel
Этот конвейер скопирует один пакет объектов. Объекты, относящиеся к этому пакету, будут копироваться параллельно.
Параметры
| Имя параметра | Description |
|---|---|
| MaxNumberOfObjectsReturnedFromLookupActivity | Чтобы избежать достижения предельного количества операций поиска в выходных данных, можно определить максимальное число объектов, возвращаемых действием поиска. В большинстве случаев значение по умолчанию не требуется изменять. |
| TopLevelPipelineName | Имя конвейера верхнего уровня. |
| TriggerName | Имя триггера. |
| CurrentSequentialNumberOfBatch | Идентификатор последовательного пакета. |
| SumOfObjectsToCopy | Общее число копируемых объектов. |
| SumOfObjectsToCopyForCurrentBatch | Число копируемых объектов в текущем пакете. |
| MainControlTableName | Имя основной таблицы управления. |
| ConnectionControlTableName | Имя таблицы управления подключениями. |
Процедуры
| Имя действия | Тип активности | Description |
|---|---|---|
| DivideOneBatchIntoMultipleGroups | ForEach | Разделение объектов из одного пакета на несколько параллельных групп, чтобы избежать достижения предела выходных данных действия поиска. |
| GetObjectsPerGroupToCopy | Подстановка | Получение объектов (таблиц и т. д.) из таблицы управления, которые должны быть скопирована в этой группе. Порядок копирования объектов после элемента TaskId в таблице управления (ORDER BY [TaskId] DESC). |
| CopyObjectsInOneGroup | Выполнение конвейера | Выполнение другого конвейера, чтобы скопировать объекты из одной группы. Объекты, относящиеся к этой группе, будут копироваться параллельно. |
MetadataDrivenCopyTask_xxx_ BottomLevel
Этот конвейер скопирует объекты из одной группы. Объекты, относящиеся к этой группе, будут копироваться параллельно.
Параметры
| Имя параметра | Description |
|---|---|
| ObjectsPerGroupToCopy | Число копируемых объектов в текущей группе. |
| ConnectionControlTableName | Имя таблицы управления подключениями. |
| windowStart | Используется для передачи текущего времени активации в конвейер, чтобы заполнить динамический путь к папке, если он настроен пользователем. |
Процедуры
| Имя действия | Тип активности | Description |
|---|---|---|
| ListObjectsFromOneGroup | ForEach | Вывод списка объектов из одной группы и их итеративная передача в нисходящие действия. |
| RouteJobsBasedOnLoadingBehavior | Switch | Проверка режима загрузки для каждого объекта. Если указано значение по умолчанию или FullLoad, выполняется полная загрузка. Если указано значение DeltaLoad, выполняется добавочная загрузка посредством столбца предела для обнаружения изменений. |
| FullLoadOneObject | Копия | Создание полного моментального снимка этого объекта и его копирование в место назначения. |
| DeltaLoadOneObject | Копия | Копирование только данных, измененных с момента последнего копирования, посредством сравнения значения в столбце предела для обнаружения изменений. |
| GetMaxWatermarkValue | Подстановка | Запрос исходного объекта, чтобы получить максимальное значение из столбца предела. |
| UpdateWatermarkColumnValue | StoreProcedure | Обратная запись нового значения предела в таблицу управления для использования в следующий раз. |
Известные ограничения
- Имя IR, тип базы данных, тип формата файла невозможно параметризовать в ADF. Например, если требуется принимать данные с сервера Oracle и сервера SQL Server, потребуются два разных параметризованных конвейера. Но два набора конвейеров могут совместно использовать одну таблицу управления.
- OPENJSON используется в сгенерированных сценариях SQL инструментом копирования данных. Если вы используете SQL Server для размещения управляющей таблицы, это должен быть SQL Server 2016 (13.x) или более поздней версии, чтобы поддерживать функцию OPENJSON.
Связанный контент
Ознакомьтесь с руководствами ниже, при работе с которыми используется инструмент копирования данных: