Качаем в локальную папку docker-compose файл. curl -LfO 'https://airflow.apache.org/docs/apache-airflow/2.5.0/docker-compose.yaml'.
Устанавливаем докер для вашей ос.
Создаем в корне проекта . env файл со следующим содержимым:export AIRFLOW_UID=50000
Запускаем команду docker-compose up. У вас будут подняты все необходимые сервисы (включая бд Postgres, redis, celery воркеры и тд). Так же у вас будут созданы необходимые директории для создания дагов и плагинов, а так-же директория с логами.
Теперь когда у вас установлен Airflow, то вы можете в нем авторизоваться. Логин и пароль по умолчанию - airflow.
Более подробные инструкции по локальной установке airflow при помощи Docker вы можете найти в документации.
В Airflow из коробки есть множество готовых примеров реализации дагов. Возьмем в качестве заготовки один из них и напишем свой простой даг с синтетическим кодом для загрузки изображений и сохранению их в условное хранилище.
В extract будем ходить в апи за json со ссылками на фотографии.
[{"id":"chb","url":"https://cdn2.thecatapi.com/images/chb.jpg","width":480,"height":360}]В transform будем преобразовывать данные в массив с названиями фотографий фото.
["chb.jpg", "chb2.jpg"]В load будем выгружать данные в условный Storage(s3, ftp и тд).
Вот так выглядит наш даг состоящий всего из 3 методов.
Давайте его запустим:
Наш даг работает и полностью подходит нам в качестве заготовки для наших будущих проектов. Как вы заметили мы никуда не сохраняем скачанные фотографии, а просто вычитываем их из респонса. Давайте попробуем поработать с хранилищем и пусть это будет S3. Так-как у нас под рукой нет бакета, то мы его сэмулируем при помощи MinIO.
Для этого давайте добавим в наш docker-compose файл еще один сервис:
s3
hostname: airflow-s3
image: minio/minio:latest
ports:
- 9000:9000
- 9001:9001
env_file:
- .env
healthcheck:
test: [ 'CMD', 'curl', 'http://airflow-s3:9000/minio/health/cluster/read' ]
start_period: 10s
timeout: 5s
interval: 3s
retries: 100
command: server --console-address :9001 /data:Добавьте в .env файл наш будущий secret_key и access_key.
export MINIO_ROOT_USER=airflow
export MINIO_ROOT_PASSWORD=airflowpassСоздадим с3 бакет. Перезапускаем наши сервисы через docker-compose restart и заходим в UI minIO. Авторизоваться нужно при помощи MINIO_ROOT_USER/PASSWORD указанных в env файле.
Теперь нужно создать новое соединение для s3 бакета. Сделать это мы можем при помощи интерфейса или командной строки. Воспользуемся UI для простоты.
В поле extra необходимо дописать: {"region_name": "us-east-1", "bucket_name": "my-s3bucket", "endpoint_url": "http://airflow-s3:9000"}
Нам необходимо добавить к нашему дагу работу с соединением и получение bucket_name из поля Extra в airflow connections.
hook = S3Hook(aws_conn_id="S3_ETL_CONN")
extra = hook.get_connection(hook.aws_conn_id).get_extra()
hook.load_bytes(r.content, key=url.split('/')[-1], bucket_name=json.loads(extra)['bucket_name'])Теперь переходим в даг и запускаем его.
Даг успешно отработал т.к в нашем minIO S3 появились фото котиков.
Источники: