Проект «Инженерия данных» для начинающих — Пакетная обработка данных

1. Введение

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

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

Тогда этот урок для вас. На этом уроке вы

Настроете Apache Airflow, AWS EMR, AWS Redshift, AWS Spectrum и AWS S3.
Изучите лучшие практики конвейера данных.
Узнаете, как выявлять точки сбоя в конвейерах данных и создавать системы, устойчивые к сбоям.
Узнаете, как спроектировать и построить конвейер данных на основе бизнес-требований.

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

2. Цель

Предположим, вы работаете в компании, которая занимается анализом поведения пользователей, собирает пользовательские данные и создает профиль пользователя. Вам поручено построить конвейер данных для заполнения таблицы user_behavior_metric. Таблица user_behavior_metric — это таблица OLAP, предназначенная для использования аналитиками, программными панелями и т. д. Она построена из

user_purchase: OLTP-таблицы с информацией о покупках пользователей.
movie_review.csv: данных, отправляемых каждый день внешним поставщиком данных.

3. Проект

Мы будем использовать Airflow для оркестрации следующих задач:

Классификация обзоров фильмов с помощью Apache Spark.
Загрузка классифицированных отзывов о фильмах в хранилище данных.
Извлечение данных о покупках пользователей из базы данных OLTP и загрузка их в хранилище данных.
Объединение данных о классификации обзоров фильмов и данных о покупках пользователей для получения параметров пользовательского поведения.

4. Настройка

4.1 Предварительное условие

Docker с не менее 4 ГБ ОЗУ и Docker Compose v1.27.0 или более поздней версии.
psql
AWS account
AWS CLI installed или configured

Для настройки инфраструктуры и базовых таблиц у нас есть скрипт setup_infra.sh. Его можно запустить в таком виде, как показано ниже.

git clone https://github.com/josephmachado/beginner_de_project.git
cd beginner_de_project
./setup_infra.sh {your-bucket-name}

Замените местозаполнитель именем вашего сегмента, например) ./setup_infra.sh sde-sample-bkt. Выбранное вами имя корзины должно быть уникальным. Если какая-либо из команд не прошла, откройте файл setup_infra.sh и вручную запустите неудачные команды.

Сценарий установки займет у вас около 10 минут, чтобы настроить необходимую инфраструктуру. Журналы установки хранятся локально в файле с именем setup.log. Войдите на www.localhost:8080, чтобы увидеть пользовательский интерфейс Airflow. И имя пользователя и пароль – airflow.

Когда вы закончите проект, не забудьте удалить инфраструктуру с помощью скрипта tear_down_infra.sh .

./tear_down_infra.sh {your-bucket-name}

4.2 Инфраструктура

Для нашего проекта у нас будет следующее

Локальные компоненты:

Веб-сервер и планировщик Apache Airflow в док-контейнерах.
База данных метаданных Apache Airflow (Postgres) в док-контейнере.

Компоненты AWS:

AWS S3 как наше озеро данных.
AWS Redshift Spectrum как наше хранилище данных.
AWS IAM, чтобы разрешить Spectrum доступ к данным в S3 во время запроса.
AWS EMR для запуска заданий Apache Spark для классификации текста.

4.3 Структура озера данных

Мы будем использовать AWS S3 в качестве озера данных. Здесь будут храниться данные из внешних систем для дальнейшей обработки. AWS S3 будет использоваться в качестве хранилища для использования с AWS Redshift Spectrum.

В нашем проекте мы будем использовать один участок памяти с несколькими папками.

raw: для хранения необработанных данных. Обозначается как Raw Area в разделе design.
stage: Обозначается как Stage Area a в разделе design.
raw: используется для хранения искровых скриптов для использования AWS EMR.

Этот шаблон позволит нам создавать разные корзины для разных сред.

4.4 Создание таблиц и конфигураций Airflow

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

таблица retail.user_purchase, определенная в pgsetup/create_user_purchase.sql в репозитории. Данные загружаются в файловую систему контейнера Postgres. Эти данные загружаются в таблицу с помощью команды COPY .
Таблица spectrum.user_purchase_staging, определяемая как имеющая данные, хранящиеся в том же месте, где находится озеро данных. Обратите внимание, что в таблице также есть раздел, определенный в файле insert_date.
Таблица spectrum.classified_movie_review, определенная как имеющая данные, хранящиеся в том же месте, где находится озеро данных.
Таблица public.user_behavior_metric — это таблица, в которую мы хотим загрузить данные.

Кроме того, скрипт также создает соединения и переменные Airflow.

redshift connection: для подключения к кластеру AWS Redshift.
postgres_default connection: для подключения к локальной базе данных Postgres.
BUCKET variable: для указания сегмента, который будет использоваться в качестве озера данных для этого конвейера.
EMR_ID variable: для отправки команд в кластер AWS EMR.

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

4.5 Затраты на инфраструктуру AWS

Мы используем

3 узла m4.xlarge типа для нашего кластера AWS EMR.
1 dc2.large для нашего кластера AWS Redshift.
1 iam role, чтобы разрешить Redshift доступ к S3.
1 S3 bucket размером около 150 МБ.

По очень грубой оценке наша установка обойдется нам в 313.91 USD в месяц. Это означает, что наша инфраструктура будет стоить примерно около 0.43 USD в час. Используйте этот калькулятор стоимости AWS для проверки своих затрат.

5. Пошаговый разбор кода

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

5.1 Загрузка данных о покупках пользователй в хранилищах данных

Чтобы загрузить данные о покупках пользователей из Postgres в AWS Redshift, мы выполняем следующие задачи.

extract_user_purchase_data: выгружает данные из Postgres в локальную файловую систему в контейнере Postgres. Эта файловая система синхронизируется между нашими локальными контейнерами Postgres и Airflow, что позволяет Airflow получить доступ к этим данным.
user_purchase_to_stage_data_lake: перемещает извлеченные данные в промежуточную область озера данных по адресу stage/user_purchase/{{ ds }}/user_purchase.csv, где ds заменяется датой запуска в формате YYYY-MM-DD . Этот ds будет служить разделом insert_date, определенным при создании таблицы.
user_purchase_stage_data_lake_to_stage_tbl: запускает запрос Redshift, чтобы сообщить таблице spectrum.user_purchase_staging о новой дате в разделе.

extract_user_purchase_data = PostgresOperator(
    dag=dag,
    task_id="extract_user_purchase_data",
    sql="./scripts/sql/unload_user_purchase.sql",
    postgres_conn_id="postgres_default",
    params={"user_purchase": "/temp/user_purchase.csv"},
    depends_on_past=True,
    wait_for_downstream=True,
)

user_purchase_to_stage_data_lake = PythonOperator(
    dag=dag,
    task_id="user_purchase_to_stage_data_lake",
    python_callable=_local_to_s3,
    op_kwargs={
        "file_name": "/temp/user_purchase.csv",
        "key": "stage/user_purchase/{{ ds }}/user_purchase.csv",
        "bucket_name": BUCKET_NAME,
        "remove_local": "true",
    },
)

user_purchase_stage_data_lake_to_stage_tbl = PythonOperator(
    dag=dag,
    task_id="user_purchase_stage_data_lake_to_stage_tbl",
    python_callable=run_redshift_external_query,
    op_kwargs={
        "qry": "alter table spectrum.user_purchase_staging add if not exists partition(insert_date='{{ ds }}') \
            location 's3://"
        + BUCKET_NAME
        + "/stage/user_purchase/{{ ds }}'",
    },
)


extract_user_purchase_data >> user_purchase_to_stage_data_lake >> user_purchase_stage_data_lake_to_stage_tbl
./scripts/sql/unload_user_purchase.sql
COPY (
       select invoice_number,
              stock_code,
              detail,
              quantity,
              invoice_date,
              unit_price,
              customer_id,
              country
       from retail.user_purchase -- we should have a date filter here to pull only required date's data
) TO '{{ params.user_purchase }}' WITH (FORMAT CSV, HEADER);
-- user_purchase will be replaced with /temp/user_purchase.csv from the params in extract_user_purchase_data task

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

5.2 Загрузка классификации обзоров фильмов в хранилище данных

Чтобы получить классифицированные данные обзоров фильмов в AWS Redshift, мы выполняем следующие задачи:

movie_review_to_raw_data_lake: : копирует локальный файл data/movie_review.csv в необработанную область озера данных.
spark_script_to_s3: копирует наш скрипт pyspark в область сценариев озера данных. Это позволяет AWS EMR ссылаться на него.
start_emr_movie_classification_script: добавляет шаги EMR, определенные в dags/scripts/emr/clean_movie_review.json, в наш кластер EMR. Эта задача добавляет в кластер 3 шага EMR, они делают следующее
1. Перемещает необработанные данные из S3 в HDFS: данные копируются из необработанной области озера данных в HDFS EMR.
2. Классифицирует обзоры фильмов: запускает сценарий pyspark классификации обзоров.
3. Перемещает секретные данные из HDFS в S3: данные копируются из HDFS EMR в промежуточную область озера данных.
wait_for_movie_classification_transformation: это задача датчика, которая ожидает завершения последнего шага (Move classified data from HDFS to S3) для завершения.

movie_review_to_raw_data_lake = PythonOperator(
    dag=dag,
    task_id="movie_review_to_raw_data_lake",
    python_callable=_local_to_s3,
    op_kwargs={
        "file_name": "/data/movie_review.csv",
        "key": "raw/movie_review/{{ ds }}/movie.csv",
        "bucket_name": BUCKET_NAME,
    },
)

spark_script_to_s3 = PythonOperator(
    dag=dag,
    task_id="spark_script_to_s3",
    python_callable=_local_to_s3,
    op_kwargs={
        "file_name": "./dags/scripts/spark/random_text_classification.py",
        "key": "scripts/random_text_classification.py",
        "bucket_name": BUCKET_NAME,
    },
)

start_emr_movie_classification_script = EmrAddStepsOperator(
    dag=dag,
    task_id="start_emr_movie_classification_script",
    job_flow_id=EMR_ID,
    aws_conn_id="aws_default",
    steps=EMR_STEPS,
    params={
        "BUCKET_NAME": BUCKET_NAME,
        "raw_movie_review": "raw/movie_review",
        "text_classifier_script": "scripts/random_text_classifier.py",
        "stage_movie_review": "stage/movie_review",
    },
    depends_on_past=True,
)

last_step = len(EMR_STEPS) - 1

wait_for_movie_classification_transformation = EmrStepSensor(
    dag=dag,
    task_id="wait_for_movie_classification_transformation",
    job_flow_id=EMR_ID,
    step_id='{{ task_instance.xcom_pull("start_emr_movie_classification_script", key="return_value")['
    + str(last_step)
    + "] }}",
    depends_on_past=True,
)

[
    movie_review_to_raw_data_lake,
    spark_script_to_s3,
] >> start_emr_movie_classification_script >> wait_for_movie_classification_transformation

5.3 Генерация метрики поведения пользователя

С данными пользовательских покупок, и классификаций фильмов в хранилище данных, мы можем получить данные для таблицы user_behavior_metric. Это делается с помощью задачи generate_user_behavior_metric. Эта задача запускает скрипт Redshift SQL для заполнения таблицы public.user_behavior_metric.

generate_user_behavior_metric = PostgresOperator(
    dag=dag,
    task_id="generate_user_behavior_metric",
    sql="scripts/sql/generate_user_behavior_metric.sql",
    postgres_conn_id="redshift",
)

end_of_data_pipeline = DummyOperator(task_id="end_of_data_pipeline", dag=dag) # dummy operator to indicate DAG complete

[
    user_purchase_stage_data_lake_to_stage_tbl,
    wait_for_movie_classification_transformation,
] >> generate_user_behavior_metric >> end_of_data_pipeline

SQL-запрос генерирует совокупные показатели на уровне клиента, используя spectrum.user_purchase_staging и spectrum.classified_movie_review.

-- scripts/sql/generate_user_behavior_metric.sql
DELETE FROM public.user_behavior_metric
WHERE insert_date = '{{ ds }}';
INSERT INTO public.user_behavior_metric (
        customerid,
        amount_spent,
        review_score,
        review_count,
        insert_date
    )

SELECT ups.customerid,
    CAST(
        SUM(ups.Quantity * ups.UnitPrice) AS DECIMAL(18, 5)
    ) AS amount_spent,
    SUM(mrcs.positive_review) AS review_score,
    count(mrcs.cid) AS review_count,
    '{{ ds }}'
FROM spectrum.user_purchase_staging ups
    JOIN (
        SELECT cid,
            CASE
                WHEN positive_review IS True THEN 1
                ELSE 0
            END AS positive_review
        FROM spectrum.classified_movie_review
        WHERE insert_date = '{{ ds }}'
    ) mrcs ON ups.customerid = mrcs.cid
WHERE ups.insert_date = '{{ ds }}'
GROUP BY ups.customerid;

Войдите на www.localhost:8080, чтобы увидеть пользовательский интерфейс Airflow. Имя пользователя и пароль – airflow. Включите DAG. На выполнение одного прогона может уйти около 10 минут.

5.4. Проверка результатов

Вы можете проверить Redshift таблицу public.user_behavior_metric из своего терминала, как показано ниже.

export REDSHIFT_HOST=$(aws redshift describe-clusters --cluster-identifier sde-batch-de-project --query 'Clusters[0].Endpoint.Address' --output text)
psql postgres://sde_user:sdeP0ssword0987@$REDSHIFT_HOST:5439/dev

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

select insert_date, count(*) as cnt from spectrum.classified_movie_review group by insert_date order by cnt desc; -- 100,000 per day
select insert_date, count(*) as cnt from spectrum.user_purchase_staging group by insert_date order by cnt desc; -- 541,908 per day
select insert_date, count(*) as cnt from public.user_behavior_metric group by insert_date order by cnt desc; -- 908 per day

Подсчеты должны совпадать. Когда вы закончите проект, не забудьте удалить инфраструктуру с помощью скрипта tear_down_infra.sh

./tear_down_infra.sh {your-bucket-name}

6. Размышления к проекту

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

Идемпотентный конвейер данных

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

Мониторинг и оповещение

Конвейер данных можно контролировать из пользовательского интерфейса Airflow. Шаги EMR можно отслеживать через пользовательский интерфейс AWS. У нас нет никаких сигналов тревоги в случае сбоя, проблем с качеством данных, зависаний задач и т.д. В реальных проектах обычно есть система мониторинга и оповещения. Некоторыми распространенными системами, используемыми для мониторинга и оповещения, являются cloud watch, datadog, или newrelic.

Контроль качества

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

Параллельные запуски

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

Изменение частоты DAG

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

Обратное заполнение

эту команду для запуска обратного заполнения. Добавьте к команде префикс docker exec -d beginner_de_project_airflow-webserver_1, чтобы запустить ее в нашем док-контейнере Airflow.

Размер данных

Будет ли конвейер данных работать успешно, если размер ваших данных увеличится в 10, 100, 1000 раз? Почему? почему нет?

7. Дальнейшее развитие проекта

Если вы заинтересованы в дальнейшей работе с этим конвейером данных, рассмотрите возможность внести свой вклад в следующее:

Модульные тесты, прогонные DAG тесты и интеграционные тесты.
Используйте API потока задач для DAG.