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

Введение

В своих предыдущих статьях я продемонстрировал, как собирать финансовые данные в автоматическом режиме и хранить их в корзинах AWS S3. Кроме того, мы провели разведочный анализ данных (EDA). Теперь давайте продолжим наше путешествие по разработке приложения для прогнозирования акций.

В следующих трех сериях статей мы разработаем полный проект машинного обучения с использованием различных методов и инструментов, включая AWS, Snowflake, Python, Snowpark и Streamlit. Конечным результатом станет веб-приложение, которое позволит нам прогнозировать будущие цены на акции Google.

Макет проекта

Вот шаги высокого уровня для этого проекта. В этой статье мы рассмотрим первые два раздела: Сбор данных и Хранение данных.

  1. Сбор данных: автоматическая загрузка исторических данных о ценах на акции в формате CSV и сохранение их в корзину AWS S3.
  2. Хранение данных. Храните данные в хранилище данных Snowflake, создав канал данных между AWS и Snowflake.
  3. Извлечение данных, предварительная обработка и EDA. Извлекайте и предварительно обрабатывайте данные с помощью Python и выполняйте базовый исследовательский анализ данных.
  4. Разработка модели машинного обучения: разработка модели машинного обучения, обучение модели на исторических данных, оценка модели и настройка гиперпараметров.
  5. Развертывание модели машинного обучения: разверните окончательную модель в Snowflake.
  6. Разработка веб-приложений. Создайте веб-приложение с помощью Streamlit и Python для взаимодействия с развернутой моделью и отображения прогнозов. И разверните финальное приложение в облаке Streamlit.

1. Сбор данных

На этом этапе мы создадим функцию AWS Lambda, которая будет запускать код Python для получения цен на акции Google и сохранять его в виде файла CSV в корзине AWS S3.

Поскольку я уже подробно рассмотрел эту тему ранее, я не буду тратить много времени на этот шаг. См. следующую статью, чтобы настроить корзину AWS S3, функцию Lambda и ежедневную автоматизацию.



Автоматизация парсинга веб-страниц с помощью Python, AWS Lambda, Amazon S3 и Amazon EventBridge CloudWatch
Редактировать описаниеblog.jovian.com



Пожалуйста, используйте следующие файлы имен, конфигураций и дополнительные файлы для этого проекта.

После успешного завершения этой настройки вы увидите ежедневный файл GOOG_ГГГГ-ММ-ДД_ЧЧММ.csv, который содержит цены на акции Google.

2. Хранение данных

На этом этапе мы узнаем об одной из ведущих платформ данных Snowflake. Мы скопируем данные, доступные в файле CSV, представленном в AWS S3, в таблицу Snowflake. Мы также создадим бесшовную интеграцию между AWS S3 и Snowflake и будем автоматически перекачивать данные с помощью Snowpipe, как только файл будет создан в AWS.

Мы рассмотрим следующие темы

  • Краткое знакомство со снежинкой
  • Настройка прав доступа на AWS
  • Создайте интеграцию с облачным хранилищем в Snowflake
  • Обновление доверительных отношений роли AWS IAM
  • Получите доступ к корзине AWS S3 из Snowflake и загрузите данные
  • Автоматизируйте конвейер данных с помощью Snowpipe
  • Текст и мониторинг автоматизации конвейера данных

Что такое Снежинка

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

Краткое руководство по Snowflake

Пожалуйста, создайте 30-дневную пробную учетную запись Snowflake, используя эту ссылку.

Вы можете войти в веб-консоль Snowflake, используя https://app.snowflake.com/. С левой стороны находится панель навигации.

Рабочие листы

Рабочие листы предоставляют простой способ написания SQL-запросов (DML и DDL), просмотра результатов и взаимодействия с ними.

Нажмите Worksheets на левой панели, а затем нажмите кнопку + Worksheet в правом верхнем углу.

Вы можете переименовать рабочий лист, как показано ниже.

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

Это укажет выбранную базу данных и схему.

Приступим к делу… (Весь код, написанный на рабочем листе, доступен здесь.)

Введите следующий SQL для создания WareHouse PRICE_PRED.

CREATE OR REPLACE WAREHOUSE PRICE_PRED WITH WAREHOUSE_SIZE='X-SMALL';

Вы можете выбрать весь оператор или просто навести курсор на оператор SQL и нажать кнопку Выполнить, расположенную в правом верхнем углу, чтобы выполнить SQL.

Вы можете перейти в раздел Администратор -> Склад на левой панели, чтобы проверить только что созданный склад.

Теперь вернитесь в StockPricePredictions и выберите склад PRICE_PRED.

После этого выполните следующие операторы, чтобы создать новую базу данных PRICE_PRED_DB и PRICE_PRED_SCHEMA.

CREATE OR REPLACE DATABASE PRICE_PRED_DB;

CREATE OR REPLACE SCHEMA PRICE_PRED_SCHEMA;

Вы можете перейти к Данные -> База данных на левой панели, чтобы проверить только что созданную базу данных и схему.

Давайте создадим таблицу для хранения цен, выполним следующий DDL на рабочем листе

create or replace table historical_prices
 (
 High float,
 Low float,
 Open float,
 Close float,
 Volume integer,
 AdjClose float,
 Date Date,
 Ticker varchar
 );

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

select * from historical_prices;

Таблица исторические_цены должна присутствовать в разделе Данные -> База данных -> PRICE_PRED_DB -> PRICE_PRED_SCHEMA -> Таблицы.

Мы наполовину закончили, стол готов. Следующим шагом является настройка интеграции между корзиной AWS S3 (где находится файл данных) и таблицами Snowflake.

Настройка прав доступа на AWS

Мы создадим необходимую IAM-политику и роли на стороне AWS.

Создание IAM-политики

Войдите в свою учетную запись AWS и перейдите на главную страницу консоли. В верхней части страницы должна быть строка поиска. Введите «IAM» в строке поиска. В заполненных результатах выберите первую службу «IAM».

  • На левой панели выберите «Политики», а затем нажмите кнопку «Создать политики», расположенную в правом верхнем углу.

  • На следующей странице выберите вкладку JSON и вставьте следующий текст. Он также доступен здесь.
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "s3:PutObject",
                "s3:GetObject",
                "s3:GetObjectVersion",
                "s3:DeleteObject",
                "s3:DeleteObjectVersion"
            ],
            "Resource": "arn:aws:s3:::stocksprice-prediction/*"
        },
        {
            "Effect": "Allow",
            "Action": [
                "s3:ListBucket",
                "s3:GetBucketLocation"
            ],
            "Resource": "arn:aws:s3:::stocksprice-prediction",
            "Condition": {
                "StringLike": {
                    "s3:prefix": [
                        "*"
                    ]
                }
            }
        }
    ]
}

  • Нажмите кнопку Далее 2 раза.
  • На странице проверки политики укажите имя политики как PricePredPolicy и соответствующее описание.
  • Затем нажмите Создать политику. PricePredPolicy должна быть доступна в списке.

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

Войдите в свою учетную запись AWS, перейдите на главную страницу консоли. В верхней части страницы должна быть строка поиска. Введите «IAM» в строке поиска. В заполненных результатах нажмите на первую услугу, то есть «IAM».

  • На левой панели нажмите «Роли», чтобы открыть страницу со списком всех доступных ролей. Нажмите кнопку «Создать роль».

  • Выберите тип доверенного объекта как аккаунт AWS.
  • выберите Эта учетная запись (номер учетной записи AWS) в разделе Учетная запись AWS.
  • В опции выберите первый флажок (требуется внешний идентификатор (рекомендуется, когда третья сторона возьмет на себя эту роль)
  • Это активирует текстовое поле «Внешний идентификатор». Сейчас введите 123, позже мы заменим этот текст.
  • Нажмите Далее.

  • На странице «Добавить разрешения» выберите созданную ранее PricePredPolicy и нажмите «Далее».

  • На странице сведений о роли укажите имя роли как PricePredRole и соответствующее описание.
  • Затем нажмите «Создать роль», PricePredRole должна быть доступна в списке.
  • Выберите PricePredRole. Откроется страница сводки. Скопируйте ARN, щелкнув значок «Копировать».
  • Сохраните эту роль ARN в блокноте для использования в будущем.

Загрузите исходный файл цен

  • Скачайте этот файл на локальный компьютер.
  • Перейдите в раздел S3 Bucket и выберите stockprice-prediction.
  • Нажмите кнопку «Загрузить», затем «Добавить файлы». Выберите загруженный ранее файл GOOG.csv.
  • Нажмите кнопку «Загрузить». GOOG.csv должен быть доступен в корзине S3.

Создайте интеграцию с облачным хранилищем в Snowflake

Вернемся к стороне Snowflake, войдем в веб-консоль Snowflake, используя https://app.snowflake.com/, и откроем рабочий лист StockPricePredictions.

Создайте объект интеграции хранилища, выполнив следующую инструкцию на рабочем листе. не забудьте установить storage_aws_role_arn в качестве текста роли ARN, скопированного в блокноте с предыдущего шага.

CREATE OR REPLACE STORAGE INTEGRATION aws_s3_integration
 type = external_stage
 storage_provider='S3'
 enabled=true
 storage_aws_role_arn='arn:aws:iam::XXXXXXXXXX:role/PricePredRole'
 storage_allowed_locations=('s3://stocksprice-prediction/');

Выполните следующую команду, чтобы получить описание интеграции хранилища.

DESC INTEGRATION AWS_S3_INTEGRATION;

Для таблицы результатов скопируйте property_value из STORAGE_AWS_IAM_USER_ARN и STORAGE_AWS_EXTERNAL_ID и сохраните их в блокноте для использования в будущем.

Обновление доверительных отношений роли AWS IAM

Это последняя часть головоломки для завершения рукопожатия между корзиной AWS S3 и Snowflake.

  • Войдите в консоль AWS, перейдите в IAM -> Роли и откройте PricePredRole.
  • перейдите на вкладку Доверительные отношения и нажмите кнопку Изменить политику доверия.
  • на вкладке «Надежные объекты» вам необходимо заменить 2 значения, как указано ниже.
  • Установите для значения STORAGE_AWS_IAM_USER_ARN (сохраненного ранее) значение AWS:
  • Установите для значения STORAGE_AWS_EXTERNAL_ID (сохраненного ранее) значение «sts:ExternalId»:
  • Нажмите кнопку Обновить политики, чтобы сохранить изменения.

Получите доступ к корзине AWS S3 из Snowflake и загрузите данные

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

войдите в Snowflake и откройте рабочий лист StockPricePredictions

Создание разрешений и форматов файлов

Выполните следующие команды на листе StockPricePredictions.

  • Предоставление разрешений роли ACCOUNTADMIN для использования интеграции
GRANT USAGE ON INTEGRATION AWS_S3_INTEGRATION TO ROLE ACCOUNTADMIN;

Создание формата файла на Snowflake для поддержки файлов .csv корзины AWS S3.

CREATE OR REPLACE FILE FORMAT csv
 type='csv'
 field_delimiter = ','
 record_delimiter = '\n'
 skip_header = 1;

Создание сцены

Это один из важных шагов, нам нужно будет создать этап, чтобы поговорить с внешней интеграцией, в данном случае AWS S3.

Выполните следующую команду на листе StockPricePredictions.

CREATE OR REPLACE STAGE price_pred_aws_stage
 storage_integration = aws_s3_integration
 file_format = csv
 url = 's3://stocksprice-prediction/';

Проверка доступных файлов на S3

Следующий запрос отобразит все доступные файлы, связанные со стадией (т. е. доступные в корзине AWS S3).

list @price_pred_aws_stage;

Загружать файлы данных из AWS S3 в таблицы Snowflake

Давайте посмотрим, как загружать данные и различные варианты.

  • Сначала мы проверим итоговые строки в таблице исторические_цены.
select count(*) from historical_prices;

Это покажет нам 0 счетчиков, так как это только что созданная таблица.

  • Выполните следующую команду, чтобы скопировать данные в исторические_цены из файла GOOG.csv, расположенного на S3.
COPY INTO historical_prices FROM @price_pred_aws_stage/GOOG.csv file_format=CSV;
  • Давайте проверим таблицу и посмотрим, загружены ли какие-либо строки.
select * from historical_prices;

вы должны увидеть в общей сложности 4667 строк в таблице.

  • Если вы попытаетесь снова запустить тот же оператор COPY INTO, Snowflake не загрузит данные, поскольку они уже были загружены.

  • Есть некоторые параметры, такие как PATTERN, ON_ERROR, FORCE, PURGE и т. д., вы можете обратиться к этой документации Snowflake, чтобы изучить эти параметры. Я также предоставлю ссылку на файл, который будет содержать все дополнительные заявления.
  • Мы будем использовать следующий оператор для выполнения копирования данных, это выполнит массовую загрузку всех доступных необработанных файлов csv из S3, и если есть какие-либо неверные данные, он пропустит эту строку и продолжит массовую загрузку.
copy into historical_prices from @price_pred_aws_stage
 file_format=csv
 pattern = '.*csv.*'
 on_error = 'Continue';
  • Теперь проверим, можем ли мы управлять корзиной S3 из Snowflake, я пытаюсь удалить файл из корзины. Выполните следующие операторы.
remove @price_pred_aws_stage/GOOG.csv;

list @price_pred_aws_stage;

Вы можете видеть, что оператор list @stage не будет отображать никаких данных, поскольку мы успешно удалили данные S3 из Snowflake.

Вы можете вернуться в консоль AWS, чтобы убедиться, что корзина S3 должна быть пустой.

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

Нам удалось установить интеграцию между Snowflake и корзиной S3, мы также загрузили данные, выполнив инструкции COPY INTO на стороне Snowflake. Теперь пришло время автоматизировать этот процесс.

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

Следующий оператор создает канал с именем get_stocks_data_pipe в схеме. Труба загружает данные со сцены в таблицу history_prices, запустите ее на SnowFlake.

create or replace pipe get_stocks_data_pipe
 auto_ingest=true
 as
 copy into historical_prices from @price_pred_aws_stage
 file_format=csv
 pattern = '.*csv.*'
 on_error = 'Continue';

Запустите следующую инструкцию, чтобы отобразить определение снежной трубы.

SHOW PIPES;

Скопируйте значение из столбца notification_channel из результата и сохраните его в блокноте для использования в будущем.

Настройка уведомлений о событиях в корзине AWS S3

Это последний этап автоматизации.

  • Теперь вернитесь в консоль AWS и откройте сегмент S3 stockprice-prediction.
  • Нажмите на вкладку «Свойства».

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

  • Укажите правильное название события, например snowpipe-event. Затем в разделе Типы событий выберите Все события создания объектов.

  • Прокрутите вниз, в разделе Назначение выберите Очередь SQS, а также выберите Ввести ARN очереди SQS.
  • в текстовом поле Очередь SQS вставьте значения notification_channel, скопированные ранее, и нажмите Сохранить изменения.

Тестирование и мониторинг автоматизации конвейера данных

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

Запуск лямбда-функции вручную

  • Перейдите в консоль AWS и откройте функцию Lambda automate-web-scraping.
  • Перейдите на вкладку «Код» и нажмите кнопку Проверить.

  • После успешного выполнения в корзине S3 должен быть создан новый файл данных GOOG_ГГГГ-ММ-ДД_ЧЧММ.csv.

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

Проверьте данные в Снежинке

Проверьте данные

Это момент истины, давайте проверим количество строк в history_prices, оно должно быть больше 4667 (поскольку файл начальной загрузки GOOG.csv содержит 4667 строк)

select count(*) from historical_prices order by date desc;

Ура! Автоматизация конвейера данных увенчалась успехом.

Проверьте историю копирования файлов

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

select * from table(information_schema.copy_history(TABLE_NAME=>'historical_prices', START_TIME=> DATEADD(hours, -10000, CURRENT_TIMESTAMP())) ) order by 3 desc;

Вы должны увидеть запись для последнего файла, созданного с помощью функции AWS Lambda.

Проверьте состояние снежной трубы

Запустите следующий SQL, чтобы проверить текущее состояние Snowpipe. Статус всегда должен быть ВЫПОЛНЯЕТСЯ, чтобы поддерживать бесшовную интеграцию.

select SYSTEM$PIPE_STATUS( 'get_stocks_data_pipe');

Код рабочего листа StockPricePredictions доступен здесь.

Функция AWS Lambda создаст файл данных о ежедневных ценах, который будет легко загружаться автоматически.

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

Рекомендации

Ссылки на некоторые полезные ссылки.

Что дальше

Это все для первой статьи. Увидимся во второй части, где мы будем играть с данными и строить модели машинного обучения.

Вот ссылки на все руководство:

Часть 1: Сбор и хранение данных (Эта статья)

Часть 2: Извлечение данных, предварительная обработка, разработка EDA и модели машинного обучения

Часть 3: Развертывание модели машинного обучения и разработка веб-приложений

Если у вас есть какие-либо вопросы или отзывы, не стесняйтесь оставлять комментарии или связаться со мной в LinkedIn. Спасибо за чтение, и если вам понравился этот пост, пожалуйста, рассмотрите возможность подписаться на меня. До следующего раза… Удачного кодирования !!

Не забудьте поставить 👏 !