Этот пост в блоге написан в соавторстве с Джеймсом Саном из Snowflake.

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

Amazon Managed Workflows для Apache Airflow (Amazon MWAA) — это служба оркестрации управляемых рабочих процессов для Apache Airflow, которую можно использовать для настройки и эксплуатации сквозных конвейеров данных в облаке в любом масштабе. Облако данных Snowflake предоставляет единый источник достоверных данных для всех ваших потребностей в данных и позволяет вашим организациям хранить, анализировать и обмениваться большими объемами данных. Сообщество Apache Airflow с открытым исходным кодом предоставляет более 1,000 готовых операторов (плагины, которые упрощают подключение к сервисам) для Apache Airflow для построения конвейеров данных.

В этом посте мы даем обзор организации вашего конвейера данных с помощью Операторы снежинки в вашей среде Amazon MWAA. Мы определяем шаги, необходимые для настройки интеграции между Amazon MWAA и Snowflake. Решение обеспечивает комплексный автоматизированный рабочий процесс, включающий прием, преобразование, анализ и потребление данных.

Обзор решения

На следующей диаграмме показана архитектура нашего решения.

Данные, используемые для преобразования и анализа, основаны на общедоступных Набор данных Citi Bike в Нью-Йорке. Данные (архивированные файлы), включающие демографические данные гонщиков и данные о поездках, копируются из общедоступного сайта Citi Bike. Простой сервис хранения Amazon (Amazon S3) в вашем аккаунте AWS. Данные распаковываются и сохраняются в другой корзине S3 (преобразованные данные могут храниться в той же корзине S3, в которую были приняты данные, но для простоты мы используем две отдельные корзины S3). Преобразованные данные затем становятся доступными Snowflake для анализа данных. Вывод запрошенных данных публикуется в Amazon Простая служба уведомлений (Amazon SNS) для потребления.

Amazon MWAA использует ориентированный ациклический граф (DAG) для запуска рабочих процессов. В этом посте мы запускаем три группы обеспечения доступности баз данных:

Следующая диаграмма иллюстрирует этот рабочий процесс.

Смотрите пост в Репо GitHub для DAG и других файлов, связанных с публикацией.

Обратите внимание, что в этом посте мы используем DAG для создания соединения Snowflake, но вы также можете создать соединение Snowflake с помощью Пользовательский интерфейс или интерфейс командной строки Airflow.

Предпосылки

Для развертывания решения вам необходимо иметь базовое представление о Snowflake и Amazon MWAA, а также выполнять следующие предварительные требования:

• Аккаунт AWS в Регион AWS, в котором поддерживается Amazon MWAA.
• Учетная запись Snowflake с учетными данными администратора. Если у вас нет учетной записи, зарегистрируйтесь 30-дневная бесплатная пробная версия. Выберите корпоративную версию Snowflake для облачной платформы AWS.
• Доступ к Amazon MWAA, Secrets Manager и Amazon SNS.
• В этом посте мы используем два сегмента S3, называемые airflow-blog-bucket-ACCOUNT_ID и citibike-tripdata-destination-ACCOUNT_ID. Amazon S3 поддерживает глобальные сегменты. Это означает, что имя каждого сегмента должно быть уникальным для всех учетных записей AWS во всех регионах внутри раздела. Если имя корзины S3 уже занято, выберите другое имя корзины S3. Создайте корзины S3 в вашем аккаунте AWS. Мы загружаем контент в корзину S3 позже в этом посте. Заменять ACCOUNT_ID с вашим собственным идентификатором учетной записи AWS или любым другим уникальным идентификатором. Детали ведра следующие:
  • airflow-blog-bucket-ACCOUNT_ID – Корзина верхнего уровня для файлов, связанных с Amazon MWAA.
  • airflow-blog-bucket-ACCOUNT_ID/requirements – Ведро, используемое для хранения requirements.txt файл, необходимый для развертывания Amazon MWAA.
  • airflow-blog-bucket-ACCOUNT_ID/dags – Бак используется для хранения файлов DAG для запуска рабочих процессов в Amazon MWAA.
  • airflow-blog-bucket-ACCOUNT_ID/dags/mwaa_snowflake_queries – Корзина, используемая для хранения SQL-запросов Snowflake.
  • citibike-tripdata-destination-ACCOUNT_ID – Корзина, используемая для хранения преобразованного набора данных.

При реализации решения в этом посте замените ссылки на airflow-blog-bucket-ACCOUNT_ID и citibike-tripdata-destination-ACCOUNT_ID с именами ваших собственных сегментов S3.

Настройка среды Amazon MWAA

Ты первый создать среду Amazon MWAA. Перед развертыванием среды загрузите большого города, подать в airflow-blog-bucket-ACCOUNT_ID/requirements Ведро S3. Файл требований основан на Amazon MWAA версии 2.6.3. Если вы тестируете другую версию Amazon MWAA, соответствующим образом обновите файл требований.

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

1. На консоли Amazon MWAA выберите Создать среду.
2. Укажите имя по вашему выбору для среды.
3. Выберите версию Airflow 2.6.3.
4. Для корзины S3 введите путь к корзине (s3:// airflow-blog-bucket-ACCOUNT_ID).
5. Для папки DAGs введите путь к папке DAGs (s3:// airflow-blog-bucket-ACCOUNT_ID/dags).
6. Для файла требований введите путь к файлу требований (s3:// airflow-blog-bucket-ACCOUNT_ID/ requirements/requirements.txt).
7. Выберите Следующая.
8. Под Networking, выберите существующее VPC или выберите Создать VPC MWAA.
9. Под Доступ к веб-серверу, выберите Публичная сеть.
10. Под Группы безопасности, покинуть Создать новую группу безопасности выбран.
11. Для того, чтобы получить Класс среды, Шифрованиеи мониторинг разделах оставьте все значения по умолчанию.
12. В Варианты конфигурации воздушного потока раздел, выбрать Добавить значение пользовательской конфигурации и настройте два значения:
    1. Поставьте Вариант конфигурации в secrets.backend и Пользовательское значение в airflow.providers.amazon.aws.secrets.secrets_manager.SecretsManagerBackend.
    2. Поставьте Вариант конфигурации в secrets.backend_kwargs и Пользовательское значение в {"connections_prefix" : "airflow/connections", "variables_prefix" : "airflow/variables"}.                     
13. В Разрешения... раздел, оставьте настройки по умолчанию и выберите Создать новую роль.
14. Выберите Следующая.
15. Когда среда Amazon MWAA станет доступной, назначьте разрешения корзины S3 Управление идентификацией и доступом AWS (IAM) роль исполнения (создается во время установки Amazon MWAA).

Это направит вас к созданной роли исполнения на консоли IAM.

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

1. В раскрывающемся меню выберите Создать встроенную политику.
2. Что касается Выберите службу, выберите S3.
3. Под Уровень доступа, укажите следующее:
   1. Расширьте Уровень списка и ListBucket.
   2. Расширьте Уровень чтения и GetObject.
   3. Расширьте Уровень записи и PutObject.
4. Под Полезные ресурсы, выберите Добавить АРН.
5. На Текст на вкладке укажите следующие ARN для доступа к корзине S3:
   1. arn:aws:s3:::airflow-blog-bucket-ACCOUNT_ID (используйте свое ведро).
   2. arn:aws:s3:::citibike-tripdata-destination-ACCOUNT_ID (используйте свое ведро).
   3. arn:aws:s3:::tripdata (это общедоступная корзина S3, в которой хранится набор данных Citi Bike; используйте ARN, как указано здесь).
6. Под Полезные ресурсы, выберите Добавить АРН.
7. На Текст на вкладке укажите следующие ARN для доступа к корзине S3:
   1. arn:aws:s3:::airflow-blog-bucket-ACCOUNT_ID/* (обязательно поставьте звездочку).
   2. arn:aws:s3:::citibike-tripdata-destination-ACCOUNT_ID /*.
   3. arn:aws:s3:::tripdata/* (это общедоступная корзина S3, в которой хранится набор данных Citi Bike, используйте ARN, как указано здесь).
8. Выберите Следующая.
9. Что касается Название политики, войти S3ReadWrite.
10. Выберите Создать политику.
11. Наконец, предоставить Amazon MWAA разрешение на доступ к секретным ключам Secrets Manager.

На этом этапе предоставляется роль выполнения Amazon MWAA для вашей среды Amazon MWAA, доступ для чтения к секретному ключу в Secrets Manager.

Роль исполнения должна иметь политики MWAA-Execution-Policy*, S3ReadWriteи SecretsManagerReadWrite прикреплен к нему.

Когда среда Amazon MWAA доступна, вы можете войти в пользовательский интерфейс Airflow из консоли Amazon MWAA, используя ссылку для Открыть пользовательский интерфейс воздушного потока.

Настройте тему SNS и подписку

Затем вы создать тему в соц. и добавить подписка к теме. Выполните следующие шаги:

1. На консоли Amazon SNS выберите Темы из панели навигации.
2. Выберите Создать тему.
3. Что касается Тип темы, выберите Стандарт.
4. Что касается Имя, войти mwaa_snowflake.
5. Остальное оставьте по умолчанию.
6. После создания темы перейдите в раздел Подписки и выберите Создать подписку.
   
7. Что касается Тема АРН, выберите mwaa_snowflake.
8. Установите протокол на Эл. адрес.
9. Что касается Конечная точка, введите свой адрес электронной почты (вы получите уведомление на свой адрес электронной почты о принятии подписки).

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

10. В консоли IAM перейдите к созданной ранее роли выполнения.
11. В раскрывающемся меню выберите Создать встроенную политику.
    
12. Что касается Выбор службы, выберите SNS.
13. Под Действия, развернуть Уровень доступа на запись и Опубликовать.
14. Под Полезные ресурсы, выберите Добавить АРН.
15. На Текст вкладке укажите ARN arn:aws:sns:<<region>>:<<our_account_ID>>:mwaa_snowflake.
16. Выберите Следующая.
17. Что касается Название политики, войти SNSPublishOnly.
18. Выберите Создать политику.

Настройка секрета Secrets Manager

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

Для создания строки подключения требуется имя хоста Snowflake и учетной записи. Войдите в свою учетную запись Snowflake и под Worksheets меню, выберите знак плюса и выберите SQL-лист. Используя лист, выполните следующие команды SQL, чтобы найти имя хоста и учетной записи.

Выполните следующий запрос для имени хоста:

WITH HOSTLIST AS (SELECT * FROM TABLE(FLATTEN(INPUT => PARSE_JSON(SYSTEM$allowlist()))))
SELECT REPLACE(VALUE:host,'"','') AS HOST
FROM HOSTLIST
WHERE VALUE:type = 'SNOWFLAKE_DEPLOYMENT_REGIONLESS';

Выполните следующий запрос для имени учетной записи:

WITH HOSTLIST AS (SELECT * FROM TABLE(FLATTEN(INPUT => PARSE_JSON(SYSTEM$allowlist()))))
SELECT REPLACE(VALUE:host,'.snowflakecomputing.com','') AS ACCOUNT
FROM HOSTLIST
WHERE VALUE:type = 'SNOWFLAKE_DEPLOYMENT';

Далее настраиваем секрет в Secrets Manager.

1. На консоли диспетчера секретов выберите Храните новый секрет.
2. Что касается Секретный тип, выберите Другой тип секрета.
3. Под Пары ключ/значение, выбрать Простой текст меню.
4. В текстовое поле введите следующий код и измените строку, чтобы она отражала информацию вашей учетной записи Snowflake:

{"host": "<<snowflake_host_name>>", "account":"<<snowflake_account_name>>","user":"<<snowflake_username>>","password":"<<snowflake_password>>","schema":"mwaa_schema","database":"mwaa_db","role":"accountadmin","warehouse":"dev_wh"}

Например:

{"host": "xxxxxx.snowflakecomputing.com", "account":"xxxxxx" ,"user":"xxxxx","password":"*****","schema":"mwaa_schema","database":"mwaa_db", "role":"accountadmin","warehouse":"dev_wh"}

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

Выберите Следующая.

5. Что касается Секретное имя, войти airflow/connections/snowflake_accountadmin.
6. Оставьте все остальные значения по умолчанию и выберите Следующая.
7. Выберите Магазин.

Обратите внимание на регион, в котором был создан секрет под Секрет АРН. Позже мы определим его как переменную в пользовательском интерфейсе Airflow.

Настройте права доступа Snowflake и роль IAM.

Затем войдите в свою учетную запись Snowflake. Убедитесь, что используемая вами учетная запись имеет доступ администратора учетной записи. Создайте лист SQL. Под рабочим листом создайте склад с именем dev_wh.

Ниже приведен пример команды SQL:

CREATE OR REPLACE WAREHOUSE dev_wh WITH WAREHOUSE_SIZE = 'xsmall' AUTO_SUSPEND = 60 INITIALLY_SUSPENDED = true AUTO_RESUME = true MIN_CLUSTER_COUNT = 1 MAX_CLUSTER_COUNT = 5;

Чтобы Snowflake мог читать данные и записывать данные в корзину S3, на которую есть ссылка на внешнем этапе (корзина S3), интеграция хранилища требуется. Выполните шаги, определенные в Вариант 1. Настройка интеграции хранилища Snowflake для доступа к Amazon S3(выполняйте только шаги 1 и 2, как описано в этом разделе).

Настройка прав доступа для корзины S3

При создании политики IAM образец кода документа политики необходим (см. следующий код), который предоставляет Snowflake необходимые разрешения для загрузки или выгрузки данных с использованием одного сегмента и пути к папке. Имя сегмента, используемое в этом сообщении: citibike-tripdata-destination-ACCOUNT_ID. Вам следует изменить его, чтобы оно отражало имя вашего сегмента.

{ "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "s3:PutObject", "s3:GetObject", "s3:GetObjectVersion", "s3:DeleteObject", "s3:DeleteObjectVersion" ], "Resource": "arn:aws:s3::: citibike-tripdata-destination-ACCOUNT_ID/*" }, { "Effect": "Allow", "Action": [ "s3:ListBucket", "s3:GetBucketLocation" ], "Resource": "arn:aws:s3:::citibike-tripdata-destination-ACCOUNT_ID" } ]
}

Создайте роль IAM

Далее вы создаете Роль IAM чтобы предоставить права доступа к корзине S3, содержащей ваши файлы данных. После создания запишите Роль ARN значение, расположенное на странице сводки роли.

Настроить переменные

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

Добавьте четыре переменные со следующими парами ключ/значение:

• Основные aws_role_arn со значением <<snowflake_aws_role_arn>> (ARN для роли mysnowflakerole отмечался ранее)
• Основные destination_bucket со значением <<bucket_name>> (в этом посте ведро используется в citibike-tripdata-destination-ACCOUNT_ID)
• Основные target_sns_arn со значением <<sns_Arn>> (тема SNS в вашем аккаунте)
• Основные sec_key_region со значением <<region_of_secret_deployment>> (регион, в котором был создан секрет в Secrets Manager)

На следующем снимке экрана показано, где найти тему ARN в SNS.

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

Поздравляем, вы выполнили все этапы настройки.

Запустите группу обеспечения доступности баз данных

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

• DAG1 (create_snowflake_connection_blog.py) – Создает соединение Snowflake в Apache Airflow. Это соединение будет использоваться для аутентификации в Snowflake. Строка подключения Snowflake хранится в диспетчере секретов, на который есть ссылка в файле DAG.
• DAG2 (create-snowflake_initial-setup_blog.py) – Создает базу данных, схему, интеграцию хранилища и сцену в Snowflake.
• DAG3 (run_mwaa_datapipeline_blog.py) – Запускает конвейер данных, который распаковывает файлы из исходной общедоступной корзины S3 и копирует их в целевую корзину S3. Следующая задача создаст в Snowflake таблицу для хранения данных. Затем данные из целевой корзины S3 будут скопированы в таблицу с использованием этапа Snowflake. После успешного копирования данных в Snowflake будет создано представление, поверх которого будут выполняться SQL-запросы.

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

1. Загрузить Группы DAG в ведро S3 airflow-blog-bucket-ACCOUNT_ID/dags.
2. Загрузить Запрос SQL файлы в корзину S3 airflow-blog-bucket-ACCOUNT_ID/dags/mwaa_snowflake_queries.
3. Войдите в пользовательский интерфейс Apache Airflow.
4. Найдите DAG1 (create_snowflake_connection_blog), снимите его с паузы и выберите значок воспроизведения, чтобы запустить его.

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

После запуска DAG1 соединение Snowflake snowflake_conn_accountadmin создается на Администратор, Коммутация .

5. Найдите и запустите DAG2 (create-snowflake_initial-setup_blog).

После запуска DAG2 в Snowflake создаются следующие объекты:

• База данных mwaa_db
• Схема mwaa_schema
• Интеграция хранилища mwaa_citibike_storage_int
• Уровень mwaa_citibike_stg

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

1. Войдите в свою учетную запись Snowflake, используя учетные данные учетной записи администратора.
2. Откройте созданный ранее лист SQL и выполните следующую команду:

DESC INTEGRATION mwaa_citibike_storage_int;

mwaa_citibike_storage_int — это имя интеграции, созданной DAG2 на предыдущем шаге.

Из выходных данных запишите значения следующих двух свойств:

• STORAGE_AWS_IAM_USER_ARN – Пользователь IAM, созданный для вашей учетной записи Snowflake.
• STORAGE_AWS_EXTERNAL_ID – Внешний идентификатор, необходимый для установления доверительных отношений.

Теперь мы предоставляем пользователю IAM Snowflake разрешения на доступ к объектам корзины.

1. На консоли IAM выберите роли в навигационной панели.
2. Выберите роль mysnowflakerole.
3. На Доверительные отношения , выберите Изменить доверительные отношения.
4. Измените документ политики с помощью DESC STORAGE INTEGRATION выходные значения, которые вы записали. Например:

{ "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "AWS": "arn:aws:iam::5xxxxxxxx:user/mgm4-s- ssca0079" }, "Action": "sts:AssumeRole", "Condition": { "StringEquals": { "sts:ExternalId": "AWSPARTNER_SFCRole=4_vsarJrupIjjJh77J9Nxxxx/j98=" } } } ]
}

Роль AWS ARN и ExternalId будет отличаться для вашей среды в зависимости от вывода DESC STORAGE INTEGRATION запрос

6. Найдите и запустите последнюю версию DAG (run_mwaa_datapipeline_blog).

В конце запуска DAG данные готовы для запроса. В этом примере запрос (поиск верхних начальных и конечных станций) выполняется как часть группы обеспечения доступности баз данных, а выходные данные можно просмотреть в Airflow. XCOM UI.

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

Другой метод визуализации результатов — непосредственно из консоли Snowflake с использованием рабочего листа Snowflake. Ниже приведен пример запроса:

SELECT START_STATION_NAME,
COUNT(START_STATION_NAME) C FROM MWAA_DB.MWAA_SCHEMA.CITIBIKE_VW GROUP BY START_STATION_NAME ORDER BY C DESC LIMIT 10;

Существуют разные способы визуализации результатов в зависимости от вашего варианта использования.

Как мы заметили, DAG1 и DAG2 необходимо запустить только один раз, чтобы настроить соединение Amazon MWAA и объекты Snowflake. DAG3 может быть считаться запускать каждую неделю или месяц. Благодаря этому решению пользователю, проверяющему данные, не нужно входить в систему Amazon MWAA или Snowflake. Вы можете запустить автоматический рабочий процесс по расписанию, который будет получать последние данные из набора данных Citi Bike и предоставлять лучшие станции отправления и назначения для данного набора данных.

Убирать

Чтобы избежать будущих расходов, удалите ресурсы AWS (пользователи и роли IAM, секреты Secrets Manager, среду Amazon MWAA, темы и подписку SNS, корзины S3) и ресурсы Snowflake (база данных, сцена, интеграция хранилища, представление, таблицы), созданные как часть этого поста.

Заключение

В этом посте мы продемонстрировали, как настроить соединение Amazon MWAA для аутентификации в Snowflake, а также в AWS, используя учетные данные пользователя AWS. Мы использовали DAG для автоматизации создания объектов Snowflake, таких как база данных, таблицы и сцена, с помощью SQL-запросов. Мы также организовали конвейер данных с помощью Amazon MWAA, который выполнял задачи, связанные с преобразованием данных, а также запросы Snowflake. Мы использовали Secrets Manager для хранения информации и учетных данных о подключении Snowflake, а также Amazon SNS для публикации выходных данных для конечного использования.

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

Чтобы узнать больше, обратитесь к следующим ресурсам:

Об авторах

Паял Сингх является архитектором партнерских решений в Amazon Web Services, специализирующимся на бессерверной платформе. Она отвечает за помощь партнерам и клиентам в модернизации и переносе их приложений на AWS.

Джеймс Сан — старший архитектор решений для партнеров в Snowflake. Он активно сотрудничает со стратегическими облачными партнерами, такими как AWS, поддерживая интеграцию продуктов и сервисов, а также разработку совместных решений. Он занимал руководящие технические должности в таких технологических компаниях, как EMC, AWS и MapR Technologies. Обладая более чем 20-летним опытом работы в области хранения и анализа данных, он также получил докторскую степень в Стэнфордском университете.

Боско Альбукерке является старшим архитектором партнерских решений в AWS и имеет более чем 20-летний опыт работы с базами данных и аналитическими продуктами от поставщиков корпоративных баз данных и облачных провайдеров. Он помогал технологическим компаниям разрабатывать и внедрять решения и продукты для анализа данных.

Манудж Арора — старший архитектор решений для стратегических клиентов в AWS. Он фокусируется на возможностях и предложениях по миграции и модернизации в AWS. Манудж работал архитектором решений для успеха партнеров в AWS в течение последних трех лет и работал с такими партнерами, как Snowflake, над созданием проектов решений, которые использовались клиентами. Вне работы он любит путешествовать, играть в теннис и исследовать новые места с семьей и друзьями.

• SEO-контент и PR-распределение. Получите усиление сегодня.
• PlatoData.Network Вертикальный генеративный ИИ. Расширьте возможности себя. Доступ здесь.
• ПлатонАйСтрим. Интеллект Web3. Расширение знаний. Доступ здесь.
• ПлатонЭСГ. Углерод, чистые технологии, Энергия, Окружающая среда, Солнечная, Управление отходами. Доступ здесь.
• ПлатонЗдоровье. Биотехнологии и клинические исследования. Доступ здесь.
• Источник: https://aws.amazon.com/blogs/big-data/use-snowflake-with-amazon-mwaa-to-orchestrate-data-pipelines/