Данные Azure Data Factory Поток для сложных преобразований данных

1.Источник

Источник определяет, откуда берутся ваши данные. Azure Data Factory поддерживает широкий спектр типов источников, включая Azure Blob Storage Gen2, Azure Data Lake Storage Gen2, Azure SQL Database, Synapse Analytics, Amazon S3, Google Cloud Storage и локальные базы данных через самоорганизующиеся среды выполнения интеграции. Каждый источник может быть настроен с деталями подключения, форматом файла (Parquet, CSV, JSON, Avro, ORC) и определением схемы. Используя Schema Drift, Data Flows может автоматически адаптироваться к изменениям в схеме источника — критическая функция для обработки полуструктурированных или развивающихся данных.

Наилучшая практика заключается в использовании форматов Parquet или Delta Lake для источника и поглотителя из-за их колоннального хранения и эффективности сжатия. Эти форматы значительно ускоряют операции чтения/записи и снижают стоимость.

2.Трансформации

ADF Data Flows предлагает богатую библиотеку мероприятий по трансформации.

• Расширители: Фильтр, сортировка и переменная строка (для операций вставки/обновления/удаления).
• Колонные модификаторы: Выберите, производную колонку, агрегируйте, окно, поворот, разворот и ранжирование.
• Множественные входы/выходы: Присоединяйтесь, смотрите, существует, союз и условный раскол.
• Модификаторы схем: Новая ветвь, Ассерт (правила качества данных) и Суррогатный ключ.

Преобразование Производная колонка особенно мощно — вы можете создавать выражения с помощью встроенного конструктора выражений, который включает функции для манипулирования строками, арифметики даты / времени, математических операций и сопоставления шаблонов (по аналогии с SQL). Например, вы можете создать новую колонку «Полное имя» путем объединения «Первое имя» и «Последнее имя» с пространством.

3. Потонуть

Sink определяет, где находятся преобразованные данные. Как и источники, поглотители могут быть любым поддерживаемым хранилищем данных. Критические настройки включают формат файла, стратегию раздела (Hash, Dynamic, Round Robin или File Name) и режим вывода (Append vs. Overwrite). Для поглотителей Delta Lake вы можете включить Слияние , Обновление или Upsert поведение, позволяя Data Flows действовать как мини-погрузчик хранилища данных.

Реализация сложных преобразований: подробный сценарий

Давайте рассмотрим реальный пример: Обогащение клиента 360. Представьте, что у вас есть три источника исходных данных:

• Профили клиентов (CSV из хранилища Blob)
• История транзакций (паркет ADLS Gen2)
• Каталог продуктов (Azure SQL Database)

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

Шаг 1: Загрузка и чистые источники

Добавить три узла источника. Для профилей клиентов использовать производную колонку для стандартизации формата «DateOfBirth» и удалить строки с нулевыми адресами электронной почты. Для транзакций отфильтровать возвращенные транзакции (где «Amount < 0»). Для каталога продукта присоедините имя категории с идентификатором категории.

Шаг 2: Присоединяйтесь к транзакциям с клиентами

Добавить преобразование Присоединиться , чтобы объединить очищенные Профили клиентов и историю транзакций на «CustomerID». Используйте внутреннее соединение, чтобы исключить клиентов без транзакций. Затем используйте преобразование Выбрать , чтобы отбросить дублирующие столбцы (например, переименовать «CustomerID» со второго входа).

Шаг 3: Совокупность на одного клиента

Подключите объединенный выход к преобразованию Aggregate.Группу по 'CustomerID' и 'CustomerName', и вычислите Sum(Amount) как TotalSpending, Count(] как TransactionCount, и Max(TransactionDate) как LastPurchaseDate.

Шаг 4: Обогащайтесь предпочтениями продукта

Используйте вторую Присоединиться, чтобы прикрепить Каталог продукта на 'ProductID' (который существует в источнике транзакции). Затем добавьте преобразование Pivot , чтобы преобразовать названия категорий в столбцы (например, электроника, одежда, дом) с подсчетом покупок в категории.

Шаг 5: Определите уровень лояльности

Добавить преобразование Производная колонка , в котором используется вложенная логика if-else для присвоения уровней лояльности: «если (TotalSpending > 10000, «Золото», if (TotalSpending > 5000, «Серебряный», «Бронзовый»)»

Шаг 6: Напишите обогащенные данные

Подключите конечный вывод к Sink, который нацелен на таблицу базы данных Azure SQL или папку Delta Lake в ADLS Gen2. Настройте раковину для использования Upsert поведения на «CustomerID», чтобы последующие запуски обновляли существующие записи вместо их дублирования.

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

Лучшие практики для высокопроизводительных потоков данных

Оптимизация производительности потока данных имеет важное значение при работе с терабайтами данных. Следуйте этим проверенным практикам:

• Используйте соответствующую размерность кластера: Для больших наборов данных выберите по меньшей мере 16-32 ядра. Для операций с интенсивной памятью (например, присоединений или агрегации) выберите Memory Optimized Compute.
• Разделите ваши данные: В настройках Источника включите обрезку разделов с помощью Параметров Раздела. Установите шаблон пути папки для чтения только соответствующих разделов.
• Минимизируйте перетасовку данных: Объединения и агрегации вызывают перетасовку операций по кластеру. Если вы можете, предварительно отфильтровать данные перед присоединением. Используйте Широковещательное присоединение для небольших таблиц поиска (например, таблица размеров 1 МБ).
• Оптимизируйте форматы файлов: Предпочитайте паркет или дельту по сравнению с CSV/JSON для источников и поглотителей. Эти столбцовые форматы уменьшают I/O и используют выталкивание предикатов.
• Уменьшить ветви преобразования: Каждый Новый Ветвь дублирует поток данных. Используйте условный сплит только тогда, когда это необходимо; в противном случае, слить условия в производных колонках.
• Использовать мониторинг потока данных: В мониторе ADF проверьте журналы выполнения потока данных на длительность этапов. Ищите долгосрочные преобразования и рассмотрите возможность их разбиения на более мелкие этапы.

Внешний ресурс: Официальное руководство Microsoft по производительности для потоков данных ADF

Мониторинг и отладка потоков данных

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

• Время обработки — время выполнения кластера Total Spark.
• Data Skew — неравномерное распределение данных по разделам, видимое на выходе стадии.
• Счета строк — неожиданные падения строк могут указывать на проблемы с фильтром или присоединением.

Для отладки используйте режим Data Flow Debug . Он работает на небольшом кластере и позволяет вам интерактивно проверять выход каждого преобразования. Для дальнейшей диагностики сложных выражений вы можете использовать преобразование Assert для проверки правил качества данных (например, «isNotNull(CustomerID)») и фиксации сбоев.

Рассмотрение вопросов безопасности

Потоки данных часто обрабатывают конфиденциальную информацию. ADF интегрируется с Azure Key Vault для хранения строк соединений и учетных данных. Всегда используйте управляемую идентификацию или основную аутентификацию службы по ключам учетной записи. Для данных в пути Data Flows используют TLS; для данных в состоянии покоя убедитесь, что ваши места хранения зашифрованы (шифрование Azure Storage включено по умолчанию). Кроме того, вы можете применять преобразования уровня столбца, такие как маскировка или хеширование в выражениях Data Flow, используя такие функции, как sha2() или substring().

Интеграция потоков данных с другими сервисами Azure

Потоки данных ADF не работают изолированно. Они могут быть организованы с другими мероприятиями ADF для создания сквозных трубопроводов:

• Выполнить трубопроводную деятельность: Запустить другой трубопровод ADF после завершения Data Flow.
• Databricks Notebook: Для расширенной аналитики или вывода ML объедините поток данных с Databricks.
• Лазурные функции: Вызовите пользовательский бессерверный код для обогащения, который требует сторонних API.
• Power BI: Поглощают преобразованные данные непосредственно в наборы данных Power BI через разъем Power BI ADF.