День открытых дверей
Всё о курсе «Data Engineer»

Проектирование DWH по модели Data Vault. На изображении (в материалах к занятию) представлена ER-модель части операционной базы данных. Для построения DWH необходимы четыре таблицы операционной системы:

- Person,
- Manager,
- Department,
- Company.


В каждой таблице отмечены первичные ключи (PK), внешние ключи (FK — связи между таблицами) и атрибуты. Предлагается построить ER-модели трех слоёв системы:

1. **Staging area (STG)** – копия данных из системы-источника. На этом этапе необходимые для анализа данные собираются из операционной базы данных без изменений. Написать SQL-запросы (не забыть про необходимые метаданные).
2. **Dimensional Data Store (DDS)** – на этом этапе над данными в STG совершаются трансформации (JOIN, добавление столбцов и т. д.), необходимые перед загрузкой в DWH. В рамках Data Vault необходимо произвести следующие операции:
- Для каждой таблицы выбрать бизнес-ключ;
- Создать хэш бизнес-ключа каждой таблицы;
- В каждую таблицу добавить технические данные (source, load_ts);
- Создать таблицы-связи (link);
- Создать таблицы-хабы;
- Создать таблицы-сателлиты.


В рамках задания предлагается просто нарисовать ER-модель, которая будет результатом трансформаций в DDS, а также написать SQL-запросы создания хабов, линков и сателлитов.




3. **Common Data Mart (CDM)** – данные, готовые для визуализации. Необходимо реализовать следующие метрики:
- Количество менеджеров в каждом департаменте;
- Количество сотрудников (не менеджеров) в каждом департаменте.


Нарисовать схему таблиц(ы) в CDM. Написать соответствующие SQL-запросы.
К каждому из трех этапов предлагается нарисовать ER-модель.




**Форма сдачи**

ER-модели можно нарисовать в draw.io (затем экспортировать как jpeg), либо в другом инструменте на ваш выбор. Изображения моделей поместить на PDF-документ (вместе с запросами SQL).




**Куда сдать ДЗ**

PDF-файл выложить в личном кабинете, на странице с домашним заданием.




**Куда и кому задавать вопросы, если они возникнут**

Прямо в чате, на странице с домашним заданием. Преподаватель ответит.




**Примерное время выполнения**

2 часа.

Hadoop в Docker. 1. Убедитесь, что у Linux VM, созданной в предыдущем ДЗ, не менее 16 GB RAM, 50 GB HDD и ОС Ubuntu не выше версии 20.
2. Запустите VM.
3. На своём компьютере добавить в файл /etc/hosts публичный IP-адрес VM под именем sandbox-hdp.hortonworks.com.

Пример:

51.250.22.171 sandbox-hdp.hortonworks.com
4. Запустите Hortonworks HDP Sandbox (https://hub.docker.com/r/hortonworks/sandbox-hdp) в Docker:
- docker run -it --rm --hostname=sandbox-hdp.hortonworks.com --privileged=true -p 2222:22 -p 8080:8080 -p 1080:1080 -p 21000:21000 -p 6080:6080 -p 50070:50070 -p 9995:9995 -p 30800:30800 -p 4200:4200 -p 8088:8088 -p 19888:19888 -p 16010:16010 -p 11000:11000 -p 8744:8744 -p 8886:8886 -p 18081:18081 -p 8443:8443 -p 10000:10000 hortonworks/sandbox-hdp:3.0.1
5. Появится приглашение «login:». Ввести «root» в качестве имени и «hadoop» в качестве пароля.
6. Появится приглашение сменить пароль пользователя root. Ввести старый пароль «hadoop», придумать и ввести новый пароль. Ввести новый пароль ещё раз для подтверждения.
7. Выполнить команду «ambari-server status» для проверки того, что Ambari запущен. В выводе команды должно присутствовать «Ambari Server running».
8. Выполнить команду «ambari-admin-password-reset», придумать и ввести пароль пользователя admin.
9. Открыть в браузере своего компьютера страницу http://sandbox-hdp.hortonworks.com:8080.
Имя пользователя: «admin», пароль — пароль, придуманный на шаге 8.
10. Проверить и запустить службы HDFS, YARN, Hive, Spark2, Zeppelin Notebook и Data Analytics Studio.
Сделать снимок экрана и приложить его к ДЗ для демонстрации выполненного задания.
11. (*) Опционально выполнить инструкции из https://github.com/hayari/data-tutorials/tree/master/tutorials/hdp/tag-based-policies-with-apache-ranger-and-apache-atlas.


Дополнительная информация о «песочнице»: https://github.com/hayari/data-tutorials/blob/master/tutorials/hdp/learning-the-ropes-of-the-hortonworks-sandbox/tutorial.md.

Анализ веб-логов с помощью ELK. https://github.com/Gorini4/elk_demo — в README репозитория содержится полная инструкция с подсказками.




1. Склонируйте репозиторий в директорию elk_demo.
2. Зайдите в эту директорию и разверните инфраструктуру, выполнив в терминале docker-compose up.
3. Отредактируйте файл clickstream.conf.
4. Загрузите данные веб-логов, выполнив команду ./load_data.sh.
5. Перейдите по адресу http://localhost:5601 и создайте отчет (dashboard), показывающий распределение запросов с разными кодами ответов (status_code) по времени.

Загрузка сырых данных. 1) Скачиваем себе образ Apache NiFi (если еще не сделали этого на занятии):

docker pull apache/nifi:latest




2) Собираем контейнер с mapping-ом 2-х портов: через который будем локально смотреть в интерфейс и через который будем общаться через curl:

docker run --name nifi -p 9090:9090 -p 8081:8081 -d -e NIFI_WEB_HTTP_PORT='9090' apache/nifi:latest




3) Создаем стартовую точку pipeline-а:

ListeHTTP-процессор.
Порт: 8081 (его мы заранее мэппили для docker-контейнера)
Base Path: loglistener




4) Второй узел. Он будет перенаправлять данные в соответствии с тем, что пришло на наш веб-сервер извне:

RouteOnContent-процессор.
Подключаем его связью "success" к ListeHTTP.
Match Requirement: "content must contain match"
Добавляем ему новое свойство errortext со значением "ERROR".
Теперь из узла способно выходить 2 потока: для flowfiles с ключевым значением (под названием errortext) и второй: unmatched




5) Проверяем работоспособность 1-го узла:

Нажимаем "Play" для ListeHTTP-процессора
Из командной строки делаем: " curl --data "someRandomText" 127.0.0.1:8081/loglistener"
Жмем Refresh (после клика ПКМ)
Видим, что в него пришли данные




6) Добавляем узел, который будет объединять несколько файлов в один:

Добавляем MergeContent-процессор.
Создаем 2 связи из RouteOnContent в MergeContent: errortext и unmatched
Настраиваем узел (ПКМ => configure):
* Scheduling -> Run Schedule: 10 sec
* Merge Strategy: Bin-Packing Algorithm
* Correlation Attribute Name: RouteOnContent.Route - этот параметр позволит нам писать файлы в разные директории в дальнейшем
* Merge Format: Binary Concatenation
* Maximum Number of Entries: 500
* Maximum Bin Age: 90s (максимальное время жизни flow-файла, после которого его принудительно выведет из pipeline)


Мы будем склеивать текстовые файлы, чтобы не плодить их в большом количестве.
Поэтому укажем формат склейки:
* Delimiter Strategy: Text
* Demarcator: открываем и в поле нажимаем Shift+Enter (это аналог \n- перевод каретки на новую строку)




7) Сохраняем полученные артефакты в облачный s3 Yandex.Cloud:

PutS3Object-процессор
* Directory: nifiHW/${RouteOnContent.Route}/${now()}.txt

Создаем бакет nifiHW через интерфейс.
У нас будет 2 папки в s3: merged и unmatched
Делаем связи от узла MergeContent вида "merged" и "unmatched"

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




8) Запускаем весь pipeline




9) Шлем много разных запросов через cli: содержащих текст ERROR внутри и нет.

Примеры:

curl --data "adbrsgbndt ERROR fevrtb" 127.0.0.1:8081/loglistener

curl --data "uiebveovne" 127.0.0.1:8081/loglistener




10) Наблюдаем за тем, что происходит внутри pipeline-а.




11) Заходим через интерфейс в s3 Yandex.Cloud и смотрим, какие новые папки и файлы появились внутри.

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




1) Создаем виртуальную машину с Apache Airflow 2.0 в Yandex Cloud.




2) Создаем «managed instance» PostgreSQL/ClickHouse/MySQL — по выбору в Yandex Cloud.
Создаем БД «analytics».




3) Добавляем наш psql в Connections через UI Airflow.




4) Выбираем один из 2-х API, с которым будем работать:
- Положение Международной Космической Станции на текущий момент времени (timestamp-latitude-longitude). Source:
http://api.open-notify.org/iss-now.json
- Курс BTC:
https://docs.coincap.io/#2a87f3d4-f61f-42d3-97e0-3a9afa41c73b

Тут нас интересует следующий endpoint: «api.coincap.io/v2/rates/bitcoin»




5) Создаем схему данных (таблицу в БД analytics) с названиями и типами полей, релевантными тому, что будем забирать из API.




6) Описываем DAG (программируем на Python) для получения данных с периодичностью 30 min.

Сам оператор для обращения к API можно выбрать любой.

Для простоты рекомендуется слать GET-запросы через python-библиотеку requests (`PythonOperator`), либо через bash (`BashOperator`) с помощью curl.




**Концептуальная схема DAG-a:** отправить запрос в API -> распарсить пришедший результат -> положить данные в БД (сделать insert в таблицу).




7) Кладем .py файл с DAG-ом в нужную директорию виртуалки с airflow.




8) Запускаем DAG тумблером в UI Airflow.




9) Отлаживаем DAG до работоспособного состояния.

В интерфейсе Airflow (облачный инстанс) есть информация о наборе успешно завершившихся DAG runs (темно-зеленые кружки) + в БД есть данные за >5 периодов времени.

В качестве проверки мы зайдем в ваш airflow webserver и отправим SQL-запрос в таблицу с данными.




10) Проверяем, что данные появились в БД.




11) Поздравляю, вы завершили ДЗ!

Сборка витрины на Spark. Инструкция:
1) Загрузить данные из [https://www.kaggle.com/datasets/AnalyzeBoston/crimes-in-boston](https://www.kaggle.com/datasets/AnalyzeBoston/crimes-in-boston).




2) Проверить данные на корректность, наличие дубликатов. Очистить.




3) Собрать витрину (агрегат по районам (поле district)) со следующими метриками:
- crimes_total — общее количество преступлений в этом районе;
- crimes_monthly — медиана числа преступлений в месяц в этом районе;
- frequent_crime_types — три самых частых crime_type за всю историю наблюдений в этом районе, объединенных через запятую с одним пробелом “, ” и расположенных в порядке убывания частоты;
- crime_type — первая часть NAME из таблицы offense_codes, разбитого по разделителю “-” (например, если NAME “BURGLARY-COMMERCIAL-ATTEMPT”, то crime_type “BURGLARY”);
- lat — широта координаты района, рассчитанная как среднее по всем широтам инцидентов;
- lng — долгота координаты района, рассчитанная как среднее по всем долготам инцидентов.




4) Сохранить витрину в один файл в формате .parquet в папке path/to/output_folder.

Подсказки:
- Функция percentile_approx может посчитать медиану.
- Конкретный месяц идентифицируется не только номером месяца, но и номером года.
- В справочнике кодов есть дубликаты. Нужно выбрать уникальные коды, взяв любое из названий.

Витрина + BI. 1. Установить Metabase
- Начало работы с интерфейсом командной строки
- Install: https://cloud.yandex.ru/docs/cli/quickstart#install
- Init profile: https://cloud.yandex.ru/docs/cli/quickstart#initialize




`yc config list` # check



- Подключиться к виртуальной машине Linux по SSH
- Создание пары ключей SSH: https://cloud.yandex.ru/docs/compute/operations/vm-connect/ssh#creating-ssh-keys






`ls ~/.ssh/id_rsa.pub # check`




- Создать виртуальную машину из публичного образа Linux https://cloud.yandex.ru/docs/compute/operations/vm-create/create-linux-vm
- проверьте путь до файла с публичным ключом




```
yc compute instance create \
--name airbyte-vm \
--ssh-key ~/.ssh/key.pub \
--create-boot-disk image-folder-id=standard-images,image-family=ubuntu-1804-lts,size=10,auto-delete=true \
--network-interface subnet-name=default-ru-central1-a,nat-ip-version=ipv4 \
--memory 2G \
--cores 2 \
--hostname airbyte-vm
```

1.


`yc compute instance list`




- Запишите публичный адрес ВМ – EXTERNAL IP

- Подключение к виртуальной машине
- https://cloud.yandex.ru/docs/compute/operations/vm-connect/ssh#vm-connect



ssh -L 3000:localhost:3000 -i ~/.ssh/key [email protected] <укажите свой EXTERNAL IP>





- If you get this:
- Permissions 0644 for '.../key.pub' are too open.
- It is required that your private key files are NOT accessible by others.
- Then change permissions: https://stackoverflow.com/questions/9270734/ssh-permissions-are-too-open-error




```
sudo chmod 600 ~/.ssh/key.pem
sudo chmod 600 ~/.ssh/key.pub
sudo chmod 755 ~/.ssh

```

- Install docker




```
sudo apt-get update
sudo apt-get install -y apt-transport-https ca-certificates curl gnupg2 software-properties-common
curl -fsSL https://download.docker.com/linux/debian/gpg | sudo apt-key add --
sudo add-apt-repository "deb [arch=amd64] https://download.docker.com/linux/debian buster stable"
sudo apt-get update
sudo apt-get install -y docker-ce docker-ce-cli containerd.io
sudo usermod -a -G docker $USER

```

- Install docker-compose




```
sudo apt-get -y install wget
sudo wget https://github.com/docker/compose/releases/download/1.26.2/docker-compose-$(uname -s)-$(uname -m) -O /usr/local/bin/docker-compose
sudo chmod +x /usr/local/bin/docker-compose
docker-compose --version

```

- Install Metabase




```
docker run -d -p 3000:3000 --name metabase metabase/metabase

```


- Done
- Access at http://localhost:3000/





2. Подключиться к источнику данных

- ClickHouse Playground: [https://clickhouse.tech/docs/en/getting-started/playground](https://clickhouse.tech/docs/en/getting-started/playground)
- [Supermarket sales](https://www.kaggle.com/datasets/aungpyaeap/supermarket-sales)
- Любой другой



3. Создать дашборд

- Нескольо видов визуализации (таблицы, графики, числа)
- Фильтры
- Блок с комментариями



4. Выслать скриншоты своего дашборда с комментариями

Проектная работа. ## Курсовой проект. Постановка задачи и требования




### Что

В качестве курсового проекта необходимо придумать дизайн и архитектуру data-driven приложения в выбранной доменной области, а затем реализовать это приложение.
Работающее приложение должно быть представлено в качестве репозитория в GitHub.




### Как

- Описать архитектуру и бизнес-требования к приложению. Для разработки схем рекомендуется использовать draw.io. Описания и схема должны быть приложены к репозиторию.
- В приложении обязательно должен быть код, который решает следующие задачи:
- Создание схемы хранилища.
- Загрузка данных из внешних источников.
- Преобразования данных, формирование слоёв хранилища.
- Оркестрация загрузок и преобразований.
- Создание дашбордов, отчётов. Визуализация результатов.