¶ Основная теория
¶ Какие данные называются большими?
Большие данные (Big Data) — это структурированные, частично структурированные или неструктурированные большие массивы данных.
Также под Big Data понимают обработку, хранение и анализ огромных объемов данных. Большие данные помогают анализировать текущее состояние бизнеса, строить прогнозы и автоматизировать рутинные процессы.
- Структурированные данные: хранятся в таблицах и реляционных базах данных
- Частично структурированные данные: данные не подходят для таблиц, но могут быть систематизированы; текстовые документы или файлы с записями о событиях.
- Неструктурированные данные: не имеют организованной структуры; аудио- и видеоматериалы, фото и другие изображения
С практической точки зрения большими называют данные, которые характеризуются следующими свойствами (3V):
- Объем (Volume): данных действительно много (терабайты, петабайты)
- Разнообразие (Variety): они имеют различную структуру, поступают из различных источников
- Скорость (Velocity): данные генерируются и обрабатываются с огромной скоростью
Большие данные — это когда хранить их все дешевле, чем определять, какая часть из них имеет ценность.
Вводятся дополнительные признаки больших данных (6V):
- Достоверность (Veracity): важны качество и точность данных, а также корректность способа, с помощью которого их получили
- Вариативность (Variability): значение одних и тех же данных может меняться с течением времени, поэтому оценивать их нужно в общем контексте
- Ценность (Value): анализ данных способен принести пользу бизнесу
Каноническими примерами больших данных:
- логи информационных систем
- информация о транзакциях в банке
- информация о покупках в интернет-магазине
- информация о перемещении объекта (GPS)
¶ Средства работы с Big Data
Средства = программы = сервисы, запускаемые в ОС на серверах.
Основные особенности:
- масштабируемость (предполагается горизонтальная за счет увеличения количества серверов)
- параллельная обработка данных
- распределенность (один сервис может быть размещен на нескольких серверах таким образом, что каждый сервер будет играть свою определенную роль в процессе функционирования сервиса)
На рисунке ниже отражены основные актуальные сегодня средства работы с большими данными:
Основные действия, совершаемые с большими данными:
- сбор (скрапперы, парсеры, датчики, сенсоры, брокеры сообщений)
- хранение (SQL и NoSQL базы данных, DWH и DL)
- обработка и анализ (Hadoop и Spark)
- представление (визуализация как частный случай)
¶ Apache Airflow
Среди средств обработки больших данных особое место занимают оркестраторы. В реальных проектах обработка данных редко состоит из одного скрипта. Обычно это сложная цепочка:
- собрать логи с серверов
- очистить / обогатить данные
- дообучить ML-модель
- сохранить результат инференса в базу данных
- отправить уведомление в Telegram, если что-то пошло не так
Позже мы узнаем, как такие процессы называются в целом
Эти задачи должны запускаться в строгом порядке. Apache Airflow — платформа, которая позволяет описывать такие цепочки, планировать их запуск и следить за выполнением.
DAG (Directed Acyclic Graph), направленный ациклический граф, — это вся цепочка задач целиком. Она имеет начало и конец, в ней нет циклов (задача не может ссылаться сама на себя). DAG определяет порядок выполнения и зависимости, но не отвечает за выполнение (это делают Scheduler и Executor).
Operator — шаблон действия. Базовые примеры:
- PythonOperator — умеет выполнять функции на Python
- BashOperator — умеет выполнять команды терминала
- EmptyOperator — ничего не делает (начало/конец)
- DockerOperator — выполняет команду внутри изолированного Docker-контейнера
- SQLExecuteQueryOperator — выполняет SQL-запросы
- Sensors — специальные операторы (FileSensor, HttpSensor и др.), ждут наступления внешнего события (появления файла, ответа API)
Task — конкретный шаг (узел) в графе, использующий определённый оператор.
XCom (Cross-Communication) — механизм передачи сообщений между задачами. Так как задачи могут выполняться на разных серверах, то они не видят переменных друг друга. XCom позволяет одной задаче положить значение в базу метаданных Airflow, а другой — забрать его.
XCom не предназначен для передачи больших данных (тяжелых файлов или таблиц
Пример простого DAG:
from airflow.decorators import dag, task
from airflow.operators.bash import BashOperator
from airflow.operators.empty import EmptyOperator
from airflow.utils.trigger_rule import TriggerRule
from datetime import datetime
import random
@dag(
dag_id='hype_dag_v3',
start_date=datetime(2026, 1, 1),
schedule=None,
catchup=False,
tags=['respect', 'airflow3']
)
def generate_hype_dag():
start = EmptyOperator(task_id='start')
@task(task_id='fetch_data')
def get_data_metric():
v = random.randint(1, 10)
print(f"Получено значение: {v}")
return v
@task.branch(task_id='check_quality_branch')
def check_quality(metric: int):
print(f"Проверка метрики: {metric}")
if metric > 5:
return 'train_model'
else:
return 'skip_processing'
train_model = BashOperator(
task_id='train_model',
bash_command='echo "Метрика респект! Запускаю обучение" && sleep 2'
)
skip_processing = BashOperator(
task_id='skip_processing',
bash_command='echo "Метрика плохая"'
)
final = EmptyOperator(
task_id='end',
trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS
)
metric_value = get_data_metric()
branch_result = check_quality(metric_value)
start >> metric_value
# зависимости
branch_result >> [train_model, skip_processing] >> final
# инициализация
generate_hype_dag()
Пример с XСom:
@task
def get_filename():
return "report_2026.csv"
# использование Jinja-шаблонизатора
process_file = BashOperator(
task_id='process',
bash_command='echo "Обрабатывается файл: {{ ti.xcom_pull(task_ids="get_filename") }}"'
)
¶ Цель работы
Самостоятельно развернуть Airflow и реализовать DAG, который автоматизирует сбор и первичную обработку данных о расписании, реагируя на наличие занятий.
¶ Задание
Установить Docker (на Windows — Docker Compose)
Развернуть локальный Airflow с использованием файлов.
Рядом с файлами создайте пустые папки
config,dags,logs,plugins
Веб-интерфейс должен быть доступен на порту 8080
Напоминание: чтобы писать код пользуйтесь плагином
Airflow Code Editor(смотритеPluginsв меню)
Реализовать задачу, которая выполняет HTTP-запрос к API расписания ОмГТУ для конкретного преподавателя "Гуненков М. Ю." за неделю с 2026.02.02 по 2026.02.08.
Откройте сайт с расписанием, найдите расписание данного преподавателя на данную неделю и используйте DevTools в браузере, чтобы обнаружить нужный URL для GET-запроса
Запрос должен выполняться асинхронно
Данные приходят в формате JSON (массив). Реализовать механизм принятия решения на основе полученных данных:
- если массив расписания не пуст, то перейти к ветке сохранения.
- если массив пустой (занятий нет), то перейти к ветке уведомления об отсутствии занятий
В ветке "есть занятия": сохранить полученный JSON в локальный файл внутри контейнера (в папку /opt/airflow/dags/data/). Имя файла должно содержать дату запуска DAG (используйте макросы Airflow).
В ветке "нет занятий": вывести в лог сообщение "На этой неделе пар у коллеги нет, можно отдыхать".
Добавьте в начало DAG проверку: пайплайн не должен начинать работу, пока в папке /opt/airflow/dags/configs/) не появится специальный файл-флаг start_process.conf (пустой). DAG должен ждать его появления.
Подсказка: проверяйте наличие папок при работе с файлами, не путайте
configиconfigs(специально так)
¶ Критерии приёмки
-
DAG отображается в веб-интерфейсе Airflow и выполняется до конца
-
при наличии данных создается JSON-файл с правильным именем, при отсутствии наблюдаем сообщение в логе
-
XCom используется для передачи данных между этапом скачивания и этапом проверки
-
реализовано ожидание появления файла-флага
Студент должен уметь запускать графы из веб-интерфейса, смотреть логи и визуальное представление, получать выходной JSON-файл
Преподаватель может попросить поменять неделю для получения расписания