Airflow DAG with PythonVirtualenvOperator Task

docker-job.py

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.operators.docker_operator import DockerOperator
from airflow.operators.python_operator import BranchPythonOperator, PythonOperator, PythonVirtualenvOperator
from airflow.operators.dummy_operator import DummyOperator
import os

default_args = {
'owner'                 : 'dimon',
'description'           : 'Use of the PythonVirtualenvOperator',
'depend_on_past'        : False,
'start_date'            : datetime(2021, 5, 1),
'email_on_failure'      : False,
'email_on_retry'        : False,
'retries'               : 1,
'retry_delay'           : timedelta(minutes=5)
}


def callable_virtualenv():
    import catboost
    return catboost.__version__

with DAG('docker_operator_dag', default_args=default_args, schedule_interval="5 * * * *", catchup=False) as dag:
    start_dag = DummyOperator(
        task_id='start_dag'
        )

    end_dag = DummyOperator(
        task_id='end_dag'
        )        

    t1 = BashOperator(
        task_id='print_current_date',
        bash_command='date'
        )


    t2 = PythonVirtualenvOperator(
        task_id="second_task",
        python_callable=callable_virtualenv,
        requirements=["catboost==0.25.0"],
        provide_context=True,
        system_site_packages=True,
    )

    t3 = PythonVirtualenvOperator(
        task_id="third_task",
        python_callable=callable_virtualenv,
        requirements=["catboost==0.26.0"],
        provide_context=True,
        system_site_packages=True,
    )

    t4 = BashOperator(
        task_id='print_hello',
        bash_command='echo "hello world"'
        )

    start_dag >> t1 
    
    t1 >> t2 >> t4
    t1 >> t3 >> t4

    t4 >> end_dag