Надежное преобразование данных с помощью потоков данных для сопоставления

ПРИМЕНИМО К: Фабрика данных Azure Azure Synapse Analytics

Совет

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

• Start a Fabric бесплатная пробная версия.
• Переход с Фабрика данных Azure на Data Factory в Microsoft Fabric

Если вы не знакомы с Фабрика данных Azure, см. статью Introduction to Фабрика данных Azure.

В этом руководстве вы будете использовать пользовательский интерфейс фабрики данных для создания конвейера, который копирует и преобразует данные из источника Azure Data Lake Storage 2-го поколения в приемник Data Lake Storage 2-го поколения (оба позволяют доступ только к выбранным сетям) с помощью потока данных сопоставления в Data Factory Managed виртуальная сеть. Шаблон конфигурации, приведенный в этом кратком руководстве, можно расширить при преобразовании данных с использованием потоков данных для сопоставления.

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

• Создали фабрику данных.
• Создайте конвейер с активностью потока данных.
• создание потока данных для сопоставления с четырьмя преобразованиями;
• тестовый запуск конвейера;
• Мониторинг активности потока данных.

Требования

• подписка Azure. Если у вас нет подписки Azure, создайте учетную запись free Azure перед началом работы.
• Azure учетная запись хранения. Вы используете Data Lake Storage как хранилища данных источника и приёмника. Если у вас нет учетной записи хранения, ознакомьтесь с Создание учетной записи хранения Azure, чтобы создать ее. Убедитесь, что получить доступ к учетной записи хранения можно только из выбранных сетей.

Файл, который мы преобразуем в этом руководстве, — moviesDB.csv, и его можно найти на GitHub. Чтобы получить файл из GitHub, скопируйте содержимое в текстовый редактор, чтобы сохранить его локально в виде файла .csv. Чтобы загрузить файл в учетную запись хранения, смотрите раздел Загрузка BLOB-объектов с помощью портала Azure. В примерах будет использоваться контейнер под названием sample-data.

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

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

1. Откройте Microsoft Edge или Google Chrome. В настоящее время только Microsoft Edge и веб-браузеры Google Chrome поддерживают пользовательский интерфейс фабрики данных.

2. В меню слева выберите Создать ресурс>Аналитика>Фабрика данных.

3. На странице Новая фабрика данных в поле Имя введите ADFTutorialDataFactory.

   Имя фабрики данных должно быть глобально уникальным. Если вы получите сообщение об ошибке, связанной со значением имени, введите другое имя для фабрики данных (например, yournameADFTutorialDataFactory). Дополнительные сведения о правилах именования артефактов в Data Factory см. в статье Правила именования Data Factory.

4. Выберите подписку Azure, в которой требуется создать фабрику данных.

5. Для группы ресурсов выполните одно из следующих действий:

   • Выберите Использовать существующуюи укажите существующую группу ресурсов в раскрывающемся списке.
   • Выберите Создать новуюи укажите имя группы ресурсов.

   Дополнительные сведения о группах ресурсов см. в статье Использовать группы ресурсов для управления ресурсами Azure.

6. В качестве версии выберите V2.

7. В поле Расположение выберите расположение фабрики данных. В раскрывающемся списке отображаются только поддерживаемые локации. Хранилища данных (например, служба хранилища Azure и База данных SQL Azure) и вычислительные ресурсы (например, Azure HDInsight), используемые фабрикой данных, могут находиться в других регионах.

8. Нажмите кнопку создания.

9. После завершения создания вы увидите уведомление в центре уведомлений. Выберите Перейти к ресурсу, чтобы открыть страницу фабрики данных.

10. Выберите Open Фабрика данных Azure Studio, чтобы запустить пользовательский интерфейс фабрики данных на отдельной вкладке.

Создайте Azure IR в управляемой виртуальной сети Data Factory

На этом шаге вы создадите Azure IR и включите управляемую виртуальную сеть Фабрики данных Azure.

1. На портале фабрики данных перейдите к Manage и выберите New для создания Azure IR.

   

2. На странице Integration runtime setup (Настройка среды выполнения интеграции) выберите, какую среду выполнения интеграции следует создать на основе требуемых возможностей. В этом руководстве выберите Azure, Self-Hosted и щелкните Continue.

3. Выберите Azure и щелкните Continue для создания среды выполнения интеграции Azure.

   

4. В разделе Virtual network configuration (Preview) (Конфигурация виртуальной сети (предварительная версия)) выберите Включить.

   

5. Нажмите кнопку создания.

Создание конвейера с помощью действия потока данных

На этом этапе вы создадите пейплайн, содержащий активность потока данных.

1. На домашней странице Фабрика данных Azure выберите Orchestrate.

   

2. На панели свойств для конвейера введите TransformMovies в поле имени конвейера.

3. В области Действия разверните меню Перемещение и преобразование. Перетащите Поток данных активность из области на полотно конвейера.

4. Во всплывающем окну Adding data flow выберите Create new data flow а затем выберите Mapping Поток данных. Завершив настройку, нажмите кнопку ОК.

   

5. Присвойте потоку данных имя TransformMovies на странице свойств.

6. В верхней строке холста конвейера переместите ползунок Поток данных debug в положение "включено". Режим отладки позволяет в интерактивном режиме тестировать логику преобразования в динамическом кластере Spark. Поток данных кластеры занимают 5–7 минут, чтобы прогреть, и пользователям рекомендуется сначала включить отладку, если планируется выполнить разработку Поток данных. Дополнительные сведения см. в статье Режим отладки.

   

Построение логики преобразования в интерфейсе потока данных

После того как вы создадите поток данных, вы автоматически перейдёте на экран редактирования потока данных. На этом шаге вы создадите поток данных, который принимает файл moviesDB.csv в Data Lake Storage и агрегирует средний рейтинг комедий с 1910 по 2000 год. Затем вы напишете этот файл обратно в Data Lake Storage.

Добавьте преобразование источника

На этом шаге вы настроили Data Lake Storage 2-го поколения в качестве источника.

1. На холсте потока данных добавьте источник, выбрав поле Добавить источник.

2. Присвойте источнику имя MoviesDB. Выберите Создать, чтобы создать исходный набор данных.

3. Выберите Azure Data Lake Storage 2-го поколения и выберите Continue.

4. Выберите DelimitedText и нажмите кнопку Продолжить.

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

6. На экране создания связанной службы назовите связанную службу Data Lake Storage 2-го поколения ADLSGen2 и укажите метод проверки подлинности. Затем введите учетные данные подключения. В этом руководстве для подключения к нашей учетной записи хранения используется ключ учетной записи.

7. Обязательно включите режим Интерактивная разработка. Его включение может занять около минуты.

   

8. Выберите Test connection (Проверить подключение). Это выйдет из строя, так как учетная запись хранения не предоставляет доступ без создания и утверждения частной конечной точки. В сообщении об ошибке вы должны увидеть ссылку для создания частной конечной точки, по которой можно перейти, чтобы создать управляемую частную конечную точку. Кроме того, можно сразу открыть вкладку Управление и выполнить инструкции из этого раздела, чтобы создать управляемую частную конечную точку.

9. Не закрывая это диалоговое окно, перейдите к учетной записи хранения.

10. Следуйте инструкциям в этом разделе, чтобы утвердить частную ссылку.

11. Вернитесь к диалоговому окну. Выберите Проверить соединение, а затем нажмите кнопку Создать, чтобы развернуть связанную службу.

12. На экране создания набора данных в поле Путь к файлу введите расположение файла. В этом руководстве файл moviesDB.csv находится в контейнере sample-data. Так как файл содержит заголовки, установите флажок Использовать первую строку как заголовок. Выберите Из подключения/хранилища, чтобы импортировать схему заголовка прямо из файла, хранящегося в хранилище. Завершив настройку, нажмите кнопку ОК.

    

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

    

Создание управляемой частной конечной точки

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

1. Перейдите на вкладку Управление.

   Примечание.

   Вкладка Управление может быть недоступна для всех экземпляров Data Factory. Если она не отображается, вы можете получить доступ к частным конечным точкам, выбрав Создание>Подключения>Частная конечная точка.

2. Перейдите в раздел Managed private endpoints (Управляемые частные конечные точки).

3. В разделе Managed private endpoints (Управляемые частные конечные точки) выберите + Создать.

   

4. Выберите плитку Azure Data Lake Storage 2-го поколения из списка и выберите Continue.

5. Введите имя созданной учетной записи хранения.

6. Нажмите кнопку создания.

7. Через несколько секунд для созданной частной ссылки отобразится состояние ожидания утверждения.

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

   

Утверждение частной ссылки в учетной записи хранения

1. В разделе Параметры для учетной записи хранения выберите Подключения частных конечных точек.

2. Установите флажок для созданной частной конечной точки и выберите Утвердить.

   

3. Добавьте описание и выберите Да.

4. Вернитесь к разделу Managed private endpoints (Управляемые частные конечные точки) на вкладке Управление для Фабрики данных.

5. Примерно через минуту отобразится сообщение об утверждении частной конечной точки.

Добавьте преобразование фильтрации

1. Выберите значок "плюс" рядом с узлом источника на холсте потока данных, чтобы добавить новое преобразование. Первое добавляемое преобразование — Фильтр.

   

2. Назовите преобразование фильтра FilterYears. Выберите поле "Выражение" рядом с полем Фильтр, чтобы открыть построитель выражений. Здесь нужно указать условие фильтрации.

   

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

   • В этом руководстве будут отфильтрованы фильмы в жанре комедии, которые вышли между 1910 и 2000 годами. Поскольку в настоящее время год является строкой, его необходимо преобразовать в целое число с помощью функции toInteger(). Используйте операторы "больше или равно" (>=) и "меньше или равно" (<=) чтобы сравнивать значения года с числами 1910 и 2000. Объедините эти выражения с помощью оператора AND (&&). Выражение будет выглядеть следующим образом:

     toInteger(year) >= 1910 && toInteger(year) <= 2000

   • Чтобы узнать, какие фильмы являются комедиями, можно использовать функцию rlike(), позволяющую найти слово "комедия" в столбце жанров. Объедините выражение rlike с выражением сравнения года и получите следующее выражение:

     toInteger(year) >= 1910 && toInteger(year) <= 2000 && rlike(genres, 'Comedy')

   • Если кластер отладки активен, можно проверить логику, нажав кнопку Обновить, чтобы увидеть результат выражения в сравнении с входными данными. Реализовать эту логику с помощью языка выражений потока данных можно разными способами.

     

   • После завершения работы с выражением нажмите кнопку Сохранить и завершить.

4. Нажмите Предварительный просмотр данных, чтобы убедиться, что фильтр работает правильно.

   

Добавление агрегатного преобразования

1. Следующее преобразование, которое необходимо добавить — это преобразование Агрегат в разделе Модификатор схемы.

   

2. Назовите агрегатное преобразование AggregateComedyRating. На вкладке Группировка выберите год из раскрывающегося списка, чтобы сгруппировать агрегаты по году, в котором вышел фильм.

   

3. Перейдите на вкладку Статистическая обработка. В левом текстовом поле присвойте столбцу имя AverageComedyRating. Выберите правое поле выражения, чтобы ввести статистическое выражение с помощью построителя выражений.

   

4. Чтобы получить среднее значение столбца Оценка, используйте агрегатную функцию avg(). Так как оценка является строковым значением, а avg() принимает числовые входные данные, необходимо преобразовать значение в число с помощью функции toInteger(). Это выражение выглядит следующим образом:

   avg(toInteger(Rating))

5. По завершении нажмите кнопку Сохранить и завершить.

   

6. Откройте вкладку Предварительный просмотр данных, чтобы просмотреть выходные данные преобразования. Обратите внимание, что здесь есть только два столбца: year и AverageComedyRating.

Добавьте преобразование, относящееся к приемнику.

1. Затем необходимо добавить преобразование Приемник под Destination.

   

2. Назовите приемник Sink. Нажмите Создать, чтобы создать набор данных приемника.

   

3. На странице New dataset выберите Azure Data Lake Storage 2-го поколения и выберите Continue.

4. На странице Выбор формата выберите DelimitedText и нажмите кнопку Продолжить.

5. Назовите набор данных приемника MoviesSink. Для связанной службы выберите ту же связанную службу ADLSGen2, которая была создана для преобразования источника. Введите выходную папку для записи данных. В этом кратком руководстве мы записываем данные в папку output в контейнере sample-data. Папка не обязательно должна существовать заранее и может быть создана динамически. Установите флажок Использовать первую строку как заголовок и выберите значение Нет для параметра Импорт схемы. Нажмите ОК.

   

Теперь создание потока данных завершено. Все готово для его запуска в конвейере.

Запуск и мониторинг потока данных

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

1. Перейдите на полотно конвейера. Нажмите кнопку Отладка, чтобы запустить отладку.

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

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

   

4. Выберите преобразование, чтобы получить подробные сведения о столбцах и разделении данных.

Если вы правильно выполнили это краткое руководство, в результирующую папку будут записаны 83 строки и 2 столбца. Вы можете убедиться в правильности данных, проверив ваше BLOB-хранилище.

Итоги

В этом руководстве вы использовали интерфейс Data Factory для создания конвейера, который копирует и преобразует данные из источника Data Lake Storage 2-го поколения в приемник Data Lake Storage 2-го поколения (оба разрешают доступ только к выбранным сетям), используя поток сопоставления данных в Data Factory Managed виртуальная сеть.