Разработка Airflow с помощью Docker

Airflow должен быть легким в работе и развитии.

Новый инструмент astro-cli от астрономов предназначен для помощи пользователям в работе с облаком и не охватывает все рабочие процессы разработки.

Это не очень хорошо работает с докером, потому что для докера недостаточно подготовки в докер-коммуникациях.

@Golang_google – лучшие гайды и туториалы для golang разработчиков

Даги, будут расположены в корне проекта в dags.

Даги, которые готовы к фиксации и пойдут в производство, хранятся в dags-production.

Вот пример структуры

/
  dags/
    include/
      helpers/
        company_name/
          module_name/

  dags-production/
    team/
      dags/
        us-east-1/
          include/
            helpers/
          *.dag
        eu-central-1/
            include/
              helpers/
            *.dag

Airflow каждую секунду пишет в свою домашнюю папку. Вот почему следует использовать файловую систему памяти.

Создание виртуальной среды

Выберем использование локальной папки airflow/, чтобы сохранить все библиотеки воздушного потока. У Airflow большое количество зависимостей.

В конце концов, Airflow — это обычный инструмент, и он должен оставаться отдельным от нашего кода.

Если у вас не установлена poetry, сделайте это сейчас.

curl -sSL <https://install.python-poetry.org> | python3 -

Затем подготовим папку airflow

mkdir airflow
echo "airflow" >> .gitignore
sudo mount -t tmpfs -o size=50m tmpfs ./airflow
poetry --directory ./airflow init --name=airflow --description=airflow --author=me --no-interaction

Затем подготовим папку airflow

Установка в виртуальной среде

Мы будем использовать poetry для инициализации новой виртуальной среды из папки ./airflow.

poetry --directory ./airflow shell
cd ..
export PROJECT_HOME=$(realpath ./)
export AIRFLOW_HOME=${PROJECT_HOME}/airflow
export AIRFLOW__CORE__LOAD_EXAMPLES=False
export AIRFLOW__CORE__DAGS_FOLDER=./dags

Теперь давайте установим Airflow, используя файл ограничений

Мы будем следовать инструкциям отсюда https://airflow.apache.org/docs/apache-airflow/stable/start/local.html 

AIRFLOW_VERSION=2.5.0
PYTHON_VERSION="$(python --version | cut -d " " -f 2 | cut -d "." -f 1-2)"

CONSTRAINT_URL="https://raw.githubusercontent.com/apache/airflow/constraints-${AIRFLOW_VERSION}/constraints-${PYTHON_VERSION}.txt"
pip install "apache-airflow==${AIRFLOW_VERSION}" --constraint "${CONSTRAINT_URL}"

pip install apache-airflow-providers-docker apache-airflow-providers-amazon
pip install apache-airflow-providers-postgres
pip install apache-airflow-providers-redis
pip install apache-airflow-providers-mysql

Мы удалим все образцы DAG. Мы хотим увидеть наших DAG гораздо быстрее.

Airflow также будет работать быстрее, потому что он не будет анализировать большое количество DAG.

Мы будем искать что-то вроде 

'/home/USER/.cache/pypoetry/virtualenvs/airflow-4vTX1qLp-py3.9'

poetry env info

echo "Show what we will delete"
find /home/guda/.cache/pypoetry/virtualenvs | grep example_dags

echo "Do actual delete"
find /home/guda/.cache/pypoetry/virtualenvs | grep example_dags | xargs rm -rf  "{}"

echo "Or do it that way"
pip show pip
find /home/guda/.cache/pypoetry/virtualenvs/bookings-bQ2s_Hyz-py3.8/lib/python3.8/site-packages | grep example_dags

Если вы забыли удалить пробные даги, будет быстрее удалить базу данных airflow и начать заново.

rm airflow/airflow.db

Наконец, запустите его

airflow standalone

В следующий раз запустите с (пароль находится в файле standalone_admin_password.txt)

Запусr Airflow в следующий раз

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

cat airflowstandalone_admin_password.txt

Мы можем создать poetry оболочку в папке ./airflow, а затем запустить airflow standalone или другие команды airflow.

Но быстрее префиксить команды airflow с помощью poetry --directory ./airflow run

export PROJECT_HOME=$(realpath ./)
export AIRFLOW_HOME=${PROJECT_HOME}/airflow
export AIRFLOW__CORE__LOAD_EXAMPLES=False
export AIRFLOW__CORE__DAGS_FOLDER=./dags

poetry --directory ./airflow run airflow standalone
poetry --directory ./airflow run airflow ...

На этом этапе вы сможете получить доступ к пустому воздушному потоку через http://127.0.0.1:8080/

Подготовка к контейнерам

DockerOperator потребуется реестр для извлечения изображений.

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

Нет понимания, почему это называется docker_default вместо image_registry_default, но это уже другая тема.

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

Работа с реестром

Извлеките и запустите реестр локально.

docker run -d -p 5000:5000 --name registry registry:2

В следующий раз у вас уже будет реестр, и вы должны запустить

docker start registry

Работа с изображениями

Когда вы создадите изображение, пометьте его тегом и нажмите на него.

docker-compose build custom_image

или

docker build .

Push & Pull

docker image tag custom_image localhost:5000/custom_image
docker push localhost:5000/custom_image
docker pull localhost:5000/custom_image

Hints & Tips для отладки локального реестра

Вам нужно указать пароль в airflow – проверьте командную строку подключения.

Пароль мог быть поддельным

docker login 127.0.0.1:5000

Когда вы указываете изображение, не забудьте установить хост примерно так:

    127.0.0.1:5000/custom_image

Список изображений

curl -X GET http://127.0.0.1:5000/v2/_catalog

Остановить реестр

docker container stop registry \
  && docker container rm -v registry

Работа с контейнерами

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

docker run -u root --entrypoint /bin/bash -ti custom_image:latest

Работа с контейнерами

Начальное значение отлично подходит для размещения в задаче justfile airflow-seed

Заполнение пользователей

poetry --directory ./airflow run airflow users delete --username admin
poetry --directory ./airflow run airflow users create --role Admin --username admin --email admin@example.com --firstname admin --lastname admin --password admin

Заполнение соединений

Вы используете direnv, верно?

poetry --directory ./airflow run \
  airflow connections add 'docker_default' \
    --conn-type 'docker' \
    --conn-login 'root' \
    --conn-host '127.0.0.1' \
    --conn-port '5000'

poetry --directory ./airflow run \
  airflow connections add 'snowflake' \
    --conn-type 'generic' \
    --conn-login '$SNOWFLAKE_USER' \
    --conn-host '$SNOWFLAKE_ACCOUNT' \
    --conn-password '$SNOWFLAKE_PASSWORD'

poetry --directory ./airflow run \
  airflow connections add 'aws' \
    --conn-type 'aws'

Заполнение переменных

poetry --directory ./airflow run \
  airflow variables set aws_region_name us-east-1

poetry --directory ./airflow run \
  airflow variables set current_aws_account NNNNNNNNNNNNNNNN

poetry --directory ./airflow run \
  airflow variables set environment production

poetry --directory ./airflow run \
  airflow variables set airflow_base_url http://localhost:8080/

Dag

Теперь, наконец, вы сможете сделать что-то подобное

from airflow import DAG
from airflow.operators.docker_operator import DockerOperator
from datetime import datetime, timedelta

# Default arguments for the DAG
default_args = {
    'owner': 'me',
    'start_date': datetime(2022, 1, 1),
    'depends_on_past': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

# Create the DAG
with DAG(
    'hello_world_dag',
    default_args=default_args,
    schedule_interval=timedelta(hours=1),
    catchup=False,
) as dag:
    # Create a task using the DockerOperator
    hello_world_task = DockerOperator(
        task_id='hello_world_task',
        image='localhost:5000/myimage:latest',
        api_version='auto',
        command='echo "hello world"',
        docker_conn_id='local_docker_registry',
    )

#Python #Docker

Источник