<p dir="auto"><strong>Description</strong><br>
Make it possible to merge multiple backfills into a single run, by extending the <code class="notranslate">start_date</code> of a single dagrun to cover a time period inclusive of all backfills.</p>
<p dir="auto"><strong>Use case / motivation</strong><br>
There are cases where running multiple backfills is less efficient than having a single run, for example where tasks in successive runs would do duplicate work.</p>
<p dir="auto"><em><strong>An example:</strong></em></p>
<ul dir="auto">
<li>We have a dag which runs every 6 hours, and processes batches of messages from the previous 6 hours by looking at the <code class="notranslate">execution_date</code> and the <code class="notranslate">next_execution_date</code> macro.</li>
<li>This dag has a task which launches a scan across a very large HBase table looking for matching rows to apply these messages to. The scan takes the same amount of time regardless of the batch size. The scan is the most time-consuming part of the dagrun (let's say it takes 3 out of 4 hours for an average dagrun).</li>
<li>An external error causes 3 successive dagruns to fail.</li>
</ul>
<p dir="auto">At this point we have 18 hours of data to catch up on. Assuming the external issue has been fixed, this would take on average 12 hours to process, meaning further delays to processing future jobs. If instead we could merge these runs into a single backfill, this would reduce the processing time from 12 hours to something like 6 hours, greatly reducing the impact of delayed processing and also resource usage on Airflow and HBase (in this case, but in general other external services).</p>
<p dir="auto">This issue of inefficient processing is one that I (and I'm sure others) have a need to solve. There are obviously other workarounds one could do but I don't think they are correct in the sense of Airflow good practices. For example:</p>
<ul dir="auto">
<li>Temporarily alter the schedule interval to cover the desired range.</li>
<li>Introduce an override in the Airflow variables to make the next run process X batches.</li>
<li>Temporarily alter the dag code.</li>
<li>Run the dag tasks manually and externally to airflow, with the desired parameters.</li>
</ul>
<p dir="auto">All of these have their own pitfalls and invariably involve some other manual intervention in Airflow to ensure the database is kept accurate and/or future runs aren't affected.</p>
<p dir="auto">If there is some other solution to this problem that I am unaware of, please let me know. I have raised this as an RFC as any change that implements this feature would touch many areas of the code base, so would require some planning.</p>
<p dir="auto">I'm using Airflow 1.10.9 with celery executor and postgres and I need to have a nested loop with the first level having 2 options and the second level going into 1800 options to loop through.</p>
<p dir="auto">I found if I go above 600 options on the second loop I get a duplicate key value error. It seems either two sessions of postgres connects or something else is triggering the duplicate entry of my first task.</p>
<p dir="auto">Here is an example DAG that exhibits the issue when you adjust the range from 10 to 1800 for <code class="notranslate">j</code>. Below should reproduce the issue.</p>
<p dir="auto">The setup is in docker and is using this docker compose file<br>
<a href="https://github.com/puckel/docker-airflow/blob/master/docker-compose-CeleryExecutor.yml">https://github.com/puckel/docker-airflow/blob/master/docker-compose-CeleryExecutor.yml</a></p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="from airflow.operators.python_operator import BranchPythonOperator
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, date, timedelta
from airflow.models import DAG
import random
import csv
from airflow.utils.dates import days_ago
from airflow.models import Variable


default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': days_ago(2),
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}



def return_branch(**kwargs):

    branches = ['branch_0,''branch_1', 'branch_2', 'branch_3', 'branch_4']

    return random.choice(branches)

with DAG(
    dag_id='branch_demo',
    default_args=default_args,
    dagrun_timeout=timedelta(hours=2),
    schedule_interval='0 11 * * *'
) as dag:
    kick_off_dag = BashOperator(task_id='run_this_first',bash_command='echo &quot;first&quot;')

    branching = BranchPythonOperator(
        task_id='branching',
        python_callable=return_branch,
        provide_context=True)

    kick_off_dag &gt;&gt; branching

    for i in range(2):
        d = BashOperator(task_id='branch_{0}'.format(i),bash_command='echo &quot;job&quot;')
        for j in range(1800):
            m = BashOperator(task_id='branch_{0}_{1}'.format(i, j),bash_command='echo &quot;done&quot;')

            d &gt;&gt; m

        branching &gt;&gt; d"><pre class="notranslate"><code class="notranslate">from airflow.operators.python_operator import BranchPythonOperator
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, date, timedelta
from airflow.models import DAG
import random
import csv
from airflow.utils.dates import days_ago
from airflow.models import Variable


default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': days_ago(2),
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}



def return_branch(**kwargs):

    branches = ['branch_0,''branch_1', 'branch_2', 'branch_3', 'branch_4']

    return random.choice(branches)

with DAG(
    dag_id='branch_demo',
    default_args=default_args,
    dagrun_timeout=timedelta(hours=2),
    schedule_interval='0 11 * * *'
) as dag:
    kick_off_dag = BashOperator(task_id='run_this_first',bash_command='echo "first"')

    branching = BranchPythonOperator(
        task_id='branching',
        python_callable=return_branch,
        provide_context=True)

    kick_off_dag &gt;&gt; branching

    for i in range(2):
        d = BashOperator(task_id='branch_{0}'.format(i),bash_command='echo "job"')
        for j in range(1800):
            m = BashOperator(task_id='branch_{0}_{1}'.format(i, j),bash_command='echo "done"')

            d &gt;&gt; m

        branching &gt;&gt; d
</code></pre></div>
<p dir="auto">Postgres log error</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="ERROR:  duplicate key value violates unique constraint &quot;task_instance_pkey&quot;
DETAIL:  Key (task_id, dag_id, execution_date)=(run_this_first, branch_demo, 2020-04-07 18:16:14.566623+00) already exists.
STATEMENT:  INSERT INTO task_instance (task_id, dag_id, execution_date, start_date, end_date, duration, state, try_number, max_tries, hostname, unixname, job_id, pool, queue, priority_weight, operator, queued_dttm, pid, executor_config) VALUES ('run_this_first', 'branch_demo', '2020-04-07T18:16:14.566623+00:00'::timestamptz, NULL, NULL, NULL, NULL, 0, 1, '', 'root', NULL, 'default_pool', 'default', 3604, NULL, NULL, NULL, '\x80047d942e'::bytea)"><pre class="notranslate"><code class="notranslate">ERROR:  duplicate key value violates unique constraint "task_instance_pkey"
DETAIL:  Key (task_id, dag_id, execution_date)=(run_this_first, branch_demo, 2020-04-07 18:16:14.566623+00) already exists.
STATEMENT:  INSERT INTO task_instance (task_id, dag_id, execution_date, start_date, end_date, duration, state, try_number, max_tries, hostname, unixname, job_id, pool, queue, priority_weight, operator, queued_dttm, pid, executor_config) VALUES ('run_this_first', 'branch_demo', '2020-04-07T18:16:14.566623+00:00'::timestamptz, NULL, NULL, NULL, NULL, 0, 1, '', 'root', NULL, 'default_pool', 'default', 3604, NULL, NULL, NULL, '\x80047d942e'::bytea)
</code></pre></div>
<p dir="auto">Traceback from scheduler logs:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="Process DagFileProcessor102313-Process:
Traceback (most recent call last):
  File &quot;/usr/local/lib/python3.7/site-packages/sqlalchemy/engine/base.py&quot;, line 1226, in _execute_context
    cursor, statement, parameters, context
  File &quot;/usr/local/lib/python3.7/site-packages/sqlalchemy/dialects/postgresql/psycopg2.py&quot;, line 854, in do_executemany
    cursor.executemany(statement, parameters)
psycopg2.errors.UniqueViolation: duplicate key value violates unique constraint &quot;task_instance_pkey&quot;
DETAIL:  Key (task_id, dag_id, execution_date)=(create_cluster, predict_model_training_combined-2, 2020-04-07 13:59:38.550927+00) already exists.
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
  File &quot;/usr/local/lib/python3.7/multiprocessing/process.py&quot;, line 297, in _bootstrap
    self.run()
  File &quot;/usr/local/lib/python3.7/multiprocessing/process.py&quot;, line 99, in run
    self._target(*self._args, **self._kwargs)
  File &quot;/usr/local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py&quot;, line 157, in _run_file_processor
    pickle_dags)
  File &quot;/usr/local/lib/python3.7/site-packages/airflow/utils/db.py&quot;, line 74, in wrapper
    return func(*args, **kwargs)
  File &quot;/usr/local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py&quot;, line 1609, in process_file
    self._process_dags(dagbag, dags, ti_keys_to_schedule)
  File &quot;/usr/local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py&quot;, line 1293, in _process_dags
    self._process_task_instances(dag, tis_out)
  File &quot;/usr/local/lib/python3.7/site-packages/airflow/utils/db.py&quot;, line 74, in wrapper
    return func(*args, **kwargs)
  File &quot;/usr/local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py&quot;, line 778, in _process_task_instances
    run.verify_integrity(session=session)
  File &quot;/usr/local/lib/python3.7/site-packages/airflow/utils/db.py&quot;, line 70, in wrapper
    return func(*args, **kwargs)
  File &quot;/usr/local/lib/python3.7/site-packages/airflow/models/dagrun.py&quot;, line 400, in verify_integrity
    session.commit()
  File &quot;/usr/local/lib/python3.7/site-packages/sqlalchemy/orm/session.py&quot;, line 1036, in commit
    self.transaction.commit()
  File &quot;/usr/local/lib/python3.7/site-packages/sqlalchemy/orm/session.py&quot;, line 503, in commit
    self._prepare_impl()
  File &quot;/usr/local/lib/python3.7/site-packages/sqlalchemy/orm/session.py&quot;, line 482, in _prepare_impl
    self.session.flush()
  File &quot;/usr/local/lib/python3.7/site-packages/sqlalchemy/orm/session.py&quot;, line 2479, in flush
    self._flush(objects)
  File &quot;/usr/local/lib/python3.7/site-packages/sqlalchemy/orm/session.py&quot;, line 2617, in _flush
    transaction.rollback(_capture_exception=True)
  File &quot;/usr/local/lib/python3.7/site-packages/sqlalchemy/util/langhelpers.py&quot;, line 68, in __exit__
    compat.reraise(exc_type, exc_value, exc_tb)
  File &quot;/usr/local/lib/python3.7/site-packages/sqlalchemy/util/compat.py&quot;, line 153, in reraise
    raise value
  File &quot;/usr/local/lib/python3.7/site-packages/sqlalchemy/orm/session.py&quot;, line 2577, in _flush
    flush_context.execute()
  File &quot;/usr/local/lib/python3.7/site-packages/sqlalchemy/orm/unitofwork.py&quot;, line 422, in execute
    rec.execute(self)
  File &quot;/usr/local/lib/python3.7/site-packages/sqlalchemy/orm/unitofwork.py&quot;, line 589, in execute
    uow,
  File &quot;/usr/local/lib/python3.7/site-packages/sqlalchemy/orm/persistence.py&quot;, line 245, in save_obj
    insert,
  File &quot;/usr/local/lib/python3.7/site-packages/sqlalchemy/orm/persistence.py&quot;, line 1084, in _emit_insert_statements
    c = cached_connections[connection].execute(statement, multiparams)
  File &quot;/usr/local/lib/python3.7/site-packages/sqlalchemy/engine/base.py&quot;, line 982, in execute
    return meth(self, multiparams, params)
  File &quot;/usr/local/lib/python3.7/site-packages/sqlalchemy/sql/elements.py&quot;, line 293, in _execute_on_connection
    return connection._execute_clauseelement(self, multiparams, params)
  File &quot;/usr/local/lib/python3.7/site-packages/sqlalchemy/engine/base.py&quot;, line 1101, in _execute_clauseelement
    distilled_params,
  File &quot;/usr/local/lib/python3.7/site-packages/sqlalchemy/engine/base.py&quot;, line 1250, in _execute_context
    e, statement, parameters, cursor, context
  File &quot;/usr/local/lib/python3.7/site-packages/sqlalchemy/engine/base.py&quot;, line 1476, in _handle_dbapi_exception
    util.raise_from_cause(sqlalchemy_exception, exc_info)
  File &quot;/usr/local/lib/python3.7/site-packages/sqlalchemy/util/compat.py&quot;, line 398, in raise_from_cause
    reraise(type(exception), exception, tb=exc_tb, cause=cause)
  File &quot;/usr/local/lib/python3.7/site-packages/sqlalchemy/util/compat.py&quot;, line 152, in reraise
    raise value.with_traceback(tb)
  File &quot;/usr/local/lib/python3.7/site-packages/sqlalchemy/engine/base.py&quot;, line 1226, in _execute_context
    cursor, statement, parameters, context
  File &quot;/usr/local/lib/python3.7/site-packages/sqlalchemy/dialects/postgresql/psycopg2.py&quot;, line 854, in do_executemany
    cursor.executemany(statement, parameters)
sqlalchemy.exc.IntegrityError: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint &quot;task_instance_pkey&quot;
DETAIL:  Key (task_id, dag_id, execution_date)=(create_cluster, predict_model_training_combined-2, 2020-04-07 13:59:38.550927+00) already exists.
[SQL: INSERT INTO task_instance (task_id, dag_id, execution_date, start_date, end_date, duration, state, try_number, max_tries, hostname, unixname, job_id, pool, queue, priority_weight, operator, queued_dttm, pid, executor_config) VALUES (%(task_id)s, %(dag_id)s, %(execution_date)s, %(start_date)s, %(end_date)s, %(duration)s, %(state)s, %(try_number)s, %(max_tries)s, %(hostname)s, %(unixname)s, %(job_id)s, %(pool)s, %(queue)s, %(priority_weight)s, %(operator)s, %(queued_dttm)s, %(pid)s, %(executor_config)s)]
[parameters: ({'task_id': 'create_cluster', 'dag_id': 'predict_model_training_combined-2', 'execution_date': datetime.datetime(2020, 4, 7, 13, 59, 38, 550927, tzinfo=&lt;TimezoneInfo [UTC, GMT, +00:00:00, STD]&gt;), 'start_date': None, 'end_date': None, 'duration': None, 'state': None, 'try_number': 0, 'max_tries': 1, 'hostname': '', 'unixname': 'root', 'job_id': None, 'pool': 'default_pool', 'queue': 'default', 'priority_weight': 7119, 'operator': None, 'queued_dttm': None, 'pid': None, 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fd653458570&gt;}, {'task_id': 'before_training', 'dag_id': 'predict_model_training_combined-2', 'execution_date': datetime.datetime(2020, 4, 7, 13, 59, 38, 550927, tzinfo=&lt;TimezoneInfo [UTC, GMT, +00:00:00, STD]&gt;), 'start_date': None, 'end_date': None, 'duration': None, 'state': None, 'try_number': 0, 'max_tries': 1, 'hostname': '', 'unixname': 'root', 'job_id': None, 'pool': 'default_pool', 'queue': 'default', 'priority_weight': 7118, 'operator': None, 'queued_dttm': None, 'pid': None, 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fd6534582d0&gt;}, {'task_id': 'after_training', 'dag_id': 'predict_model_training_combined-2', 'execution_date': datetime.datetime(2020, 4, 7, 13, 59, 38, 550927, tzinfo=&lt;TimezoneInfo [UTC, GMT, +00:00:00, STD]&gt;), 'start_date': None, 'end_date': None, 'duration': None, 'state': None, 'try_number': 0, 'max_tries': 1, 'hostname': '', 'unixname': 'root', 'job_id': None, 'pool': 'default_pool', 'queue': 'default', 'priority_weight': 1, 'operator': None, 'queued_dttm': None, 'pid': None, 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fd653458540&gt;}, {'task_id': 'remove_cluster', 'dag_id': 'predict_model_training_combined-2', 'execution_date': datetime.datetime(2020, 4, 7, 13, 59, 38, 550927, tzinfo=&lt;TimezoneInfo [UTC, GMT, +00:00:00, STD]&gt;), 'start_date': None, 'end_date': None, 'duration': None, 'state': None, 'try_number': 0, 'max_tries': 1, 'hostname': '', 'unixname': 'root', 'job_id': None, 'pool': 'default_pool', 'queue': 'default', 'priority_weight': 1, 'operator': None, 'queued_dttm': None, 'pid': None, 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fd653458330&gt;}, {'task_id': 'train_model_AGGH_eon', 'dag_id': 'predict_model_training_combined-2', 'execution_date': datetime.datetime(2020, 4, 7, 13, 59, 38, 550927, tzinfo=&lt;TimezoneInfo [UTC, GMT, +00:00:00, STD]&gt;), 'start_date': None, 'end_date': None, 'duration': None, 'state': None, 'try_number': 0, 'max_tries': 1, 'hostname': '', 'unixname': 'root', 'job_id': None, 'pool': 'default_pool', 'queue': 'default', 'priority_weight': 3, 'operator': None, 'queued_dttm': None, 'pid': None, 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fd653458180&gt;}, {'task_id': 'watch_training_AGGH_eon', 'dag_id': 'predict_model_training_combined-2', 'execution_date': datetime.datetime(2020, 4, 7, 13, 59, 38, 550927, tzinfo=&lt;TimezoneInfo [UTC, GMT, +00:00:00, STD]&gt;), 'start_date': None, 'end_date': None, 'duration': None, 'state': None, 'try_number': 0, 'max_tries': 1, 'hostname': '', 'unixname': 'root', 'job_id': None, 'pool': 'default_pool', 'queue': 'default', 'priority_weight': 2, 'operator': None, 'queued_dttm': None, 'pid': None, 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fd653458360&gt;}, {'task_id': 'train_model_AGGH_ein', 'dag_id': 'predict_model_training_combined-2', 'execution_date': datetime.datetime(2020, 4, 7, 13, 59, 38, 550927, tzinfo=&lt;TimezoneInfo [UTC, GMT, +00:00:00, STD]&gt;), 'start_date': None, 'end_date': None, 'duration': None, 'state': None, 'try_number': 0, 'max_tries': 1, 'hostname': '', 'unixname': 'root', 'job_id': None, 'pool': 'default_pool', 'queue': 'default', 'priority_weight': 3, 'operator': None, 'queued_dttm': None, 'pid': None, 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fd653458630&gt;}, {'task_id': 'watch_training_AGGH_ein', 'dag_id': 'predict_model_training_combined-2', 'execution_date': datetime.datetime(2020, 4, 7, 13, 59, 38, 550927, tzinfo=&lt;TimezoneInfo [UTC, GMT, +00:00:00, STD]&gt;), 'start_date': None, 'end_date': None, 'duration': None, 'state': None, 'try_number': 0, 'max_tries': 1, 'hostname': '', 'unixname': 'root', 'job_id': None, 'pool': 'default_pool', 'queue': 'default', 'priority_weight': 2, 'operator': None, 'queued_dttm': None, 'pid': None, 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fd6534586f0&gt;}  ... displaying 10 of 7120 total bound parameter sets ...  {'task_id': 'train_model_ZYYJ_ein', 'dag_id': 'predict_model_training_combined-2', 'execution_date': datetime.datetime(2020, 4, 7, 13, 59, 38, 550927, tzinfo=&lt;TimezoneInfo [UTC, GMT, +00:00:00, STD]&gt;), 'start_date': None, 'end_date': None, 'duration': None, 'state': None, 'try_number': 0, 'max_tries': 1, 'hostname': '', 'unixname': 'root', 'job_id': None, 'pool': 'default_pool', 'queue': 'default', 'priority_weight': 3, 'operator': None, 'queued_dttm': None, 'pid': None, 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fd652758660&gt;}, {'task_id': 'watch_training_ZYYJ_ein', 'dag_id': 'predict_model_training_combined-2', 'execution_date': datetime.datetime(2020, 4, 7, 13, 59, 38, 550927, tzinfo=&lt;TimezoneInfo [UTC, GMT, +00:00:00, STD]&gt;), 'start_date': None, 'end_date': None, 'duration': None, 'state': None, 'try_number': 0, 'max_tries': 1, 'hostname': '', 'unixname': 'root', 'job_id': None, 'pool': 'default_pool', 'queue': 'default', 'priority_weight': 2, 'operator': None, 'queued_dttm': None, 'pid': None, 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fd6527586f0&gt;})]
(Background on this error at: http://sqlalche.me/e/gkpj)
Process DagFileProcessor104982-Process:
Traceback (most recent call last):
  File &quot;/usr/local/lib/python3.7/site-packages/sqlalchemy/engine/base.py&quot;, line 1226, in _execute_context
    cursor, statement, parameters, context
  File &quot;/usr/local/lib/python3.7/site-packages/sqlalchemy/dialects/postgresql/psycopg2.py&quot;, line 854, in do_executemany
    cursor.executemany(statement, parameters)
psycopg2.errors.UniqueViolation: duplicate key value violates unique constraint &quot;task_instance_pkey&quot;
DETAIL:  Key (task_id, dag_id, execution_date)=(create_cluster, predict_model_training_combined-4, 2020-04-07 14:41:02.111727+00) already exists."><pre class="notranslate"><code class="notranslate">Process DagFileProcessor102313-Process:
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 1226, in _execute_context
    cursor, statement, parameters, context
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/dialects/postgresql/psycopg2.py", line 854, in do_executemany
    cursor.executemany(statement, parameters)
psycopg2.errors.UniqueViolation: duplicate key value violates unique constraint "task_instance_pkey"
DETAIL:  Key (task_id, dag_id, execution_date)=(create_cluster, predict_model_training_combined-2, 2020-04-07 13:59:38.550927+00) already exists.
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/multiprocessing/process.py", line 297, in _bootstrap
    self.run()
  File "/usr/local/lib/python3.7/multiprocessing/process.py", line 99, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py", line 157, in _run_file_processor
    pickle_dags)
  File "/usr/local/lib/python3.7/site-packages/airflow/utils/db.py", line 74, in wrapper
    return func(*args, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py", line 1609, in process_file
    self._process_dags(dagbag, dags, ti_keys_to_schedule)
  File "/usr/local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py", line 1293, in _process_dags
    self._process_task_instances(dag, tis_out)
  File "/usr/local/lib/python3.7/site-packages/airflow/utils/db.py", line 74, in wrapper
    return func(*args, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py", line 778, in _process_task_instances
    run.verify_integrity(session=session)
  File "/usr/local/lib/python3.7/site-packages/airflow/utils/db.py", line 70, in wrapper
    return func(*args, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/airflow/models/dagrun.py", line 400, in verify_integrity
    session.commit()
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/orm/session.py", line 1036, in commit
    self.transaction.commit()
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/orm/session.py", line 503, in commit
    self._prepare_impl()
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/orm/session.py", line 482, in _prepare_impl
    self.session.flush()
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/orm/session.py", line 2479, in flush
    self._flush(objects)
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/orm/session.py", line 2617, in _flush
    transaction.rollback(_capture_exception=True)
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/util/langhelpers.py", line 68, in __exit__
    compat.reraise(exc_type, exc_value, exc_tb)
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/util/compat.py", line 153, in reraise
    raise value
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/orm/session.py", line 2577, in _flush
    flush_context.execute()
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/orm/unitofwork.py", line 422, in execute
    rec.execute(self)
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/orm/unitofwork.py", line 589, in execute
    uow,
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/orm/persistence.py", line 245, in save_obj
    insert,
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/orm/persistence.py", line 1084, in _emit_insert_statements
    c = cached_connections[connection].execute(statement, multiparams)
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 982, in execute
    return meth(self, multiparams, params)
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/sql/elements.py", line 293, in _execute_on_connection
    return connection._execute_clauseelement(self, multiparams, params)
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 1101, in _execute_clauseelement
    distilled_params,
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 1250, in _execute_context
    e, statement, parameters, cursor, context
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 1476, in _handle_dbapi_exception
    util.raise_from_cause(sqlalchemy_exception, exc_info)
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/util/compat.py", line 398, in raise_from_cause
    reraise(type(exception), exception, tb=exc_tb, cause=cause)
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/util/compat.py", line 152, in reraise
    raise value.with_traceback(tb)
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 1226, in _execute_context
    cursor, statement, parameters, context
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/dialects/postgresql/psycopg2.py", line 854, in do_executemany
    cursor.executemany(statement, parameters)
sqlalchemy.exc.IntegrityError: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "task_instance_pkey"
DETAIL:  Key (task_id, dag_id, execution_date)=(create_cluster, predict_model_training_combined-2, 2020-04-07 13:59:38.550927+00) already exists.
[SQL: INSERT INTO task_instance (task_id, dag_id, execution_date, start_date, end_date, duration, state, try_number, max_tries, hostname, unixname, job_id, pool, queue, priority_weight, operator, queued_dttm, pid, executor_config) VALUES (%(task_id)s, %(dag_id)s, %(execution_date)s, %(start_date)s, %(end_date)s, %(duration)s, %(state)s, %(try_number)s, %(max_tries)s, %(hostname)s, %(unixname)s, %(job_id)s, %(pool)s, %(queue)s, %(priority_weight)s, %(operator)s, %(queued_dttm)s, %(pid)s, %(executor_config)s)]
[parameters: ({'task_id': 'create_cluster', 'dag_id': 'predict_model_training_combined-2', 'execution_date': datetime.datetime(2020, 4, 7, 13, 59, 38, 550927, tzinfo=&lt;TimezoneInfo [UTC, GMT, +00:00:00, STD]&gt;), 'start_date': None, 'end_date': None, 'duration': None, 'state': None, 'try_number': 0, 'max_tries': 1, 'hostname': '', 'unixname': 'root', 'job_id': None, 'pool': 'default_pool', 'queue': 'default', 'priority_weight': 7119, 'operator': None, 'queued_dttm': None, 'pid': None, 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fd653458570&gt;}, {'task_id': 'before_training', 'dag_id': 'predict_model_training_combined-2', 'execution_date': datetime.datetime(2020, 4, 7, 13, 59, 38, 550927, tzinfo=&lt;TimezoneInfo [UTC, GMT, +00:00:00, STD]&gt;), 'start_date': None, 'end_date': None, 'duration': None, 'state': None, 'try_number': 0, 'max_tries': 1, 'hostname': '', 'unixname': 'root', 'job_id': None, 'pool': 'default_pool', 'queue': 'default', 'priority_weight': 7118, 'operator': None, 'queued_dttm': None, 'pid': None, 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fd6534582d0&gt;}, {'task_id': 'after_training', 'dag_id': 'predict_model_training_combined-2', 'execution_date': datetime.datetime(2020, 4, 7, 13, 59, 38, 550927, tzinfo=&lt;TimezoneInfo [UTC, GMT, +00:00:00, STD]&gt;), 'start_date': None, 'end_date': None, 'duration': None, 'state': None, 'try_number': 0, 'max_tries': 1, 'hostname': '', 'unixname': 'root', 'job_id': None, 'pool': 'default_pool', 'queue': 'default', 'priority_weight': 1, 'operator': None, 'queued_dttm': None, 'pid': None, 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fd653458540&gt;}, {'task_id': 'remove_cluster', 'dag_id': 'predict_model_training_combined-2', 'execution_date': datetime.datetime(2020, 4, 7, 13, 59, 38, 550927, tzinfo=&lt;TimezoneInfo [UTC, GMT, +00:00:00, STD]&gt;), 'start_date': None, 'end_date': None, 'duration': None, 'state': None, 'try_number': 0, 'max_tries': 1, 'hostname': '', 'unixname': 'root', 'job_id': None, 'pool': 'default_pool', 'queue': 'default', 'priority_weight': 1, 'operator': None, 'queued_dttm': None, 'pid': None, 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fd653458330&gt;}, {'task_id': 'train_model_AGGH_eon', 'dag_id': 'predict_model_training_combined-2', 'execution_date': datetime.datetime(2020, 4, 7, 13, 59, 38, 550927, tzinfo=&lt;TimezoneInfo [UTC, GMT, +00:00:00, STD]&gt;), 'start_date': None, 'end_date': None, 'duration': None, 'state': None, 'try_number': 0, 'max_tries': 1, 'hostname': '', 'unixname': 'root', 'job_id': None, 'pool': 'default_pool', 'queue': 'default', 'priority_weight': 3, 'operator': None, 'queued_dttm': None, 'pid': None, 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fd653458180&gt;}, {'task_id': 'watch_training_AGGH_eon', 'dag_id': 'predict_model_training_combined-2', 'execution_date': datetime.datetime(2020, 4, 7, 13, 59, 38, 550927, tzinfo=&lt;TimezoneInfo [UTC, GMT, +00:00:00, STD]&gt;), 'start_date': None, 'end_date': None, 'duration': None, 'state': None, 'try_number': 0, 'max_tries': 1, 'hostname': '', 'unixname': 'root', 'job_id': None, 'pool': 'default_pool', 'queue': 'default', 'priority_weight': 2, 'operator': None, 'queued_dttm': None, 'pid': None, 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fd653458360&gt;}, {'task_id': 'train_model_AGGH_ein', 'dag_id': 'predict_model_training_combined-2', 'execution_date': datetime.datetime(2020, 4, 7, 13, 59, 38, 550927, tzinfo=&lt;TimezoneInfo [UTC, GMT, +00:00:00, STD]&gt;), 'start_date': None, 'end_date': None, 'duration': None, 'state': None, 'try_number': 0, 'max_tries': 1, 'hostname': '', 'unixname': 'root', 'job_id': None, 'pool': 'default_pool', 'queue': 'default', 'priority_weight': 3, 'operator': None, 'queued_dttm': None, 'pid': None, 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fd653458630&gt;}, {'task_id': 'watch_training_AGGH_ein', 'dag_id': 'predict_model_training_combined-2', 'execution_date': datetime.datetime(2020, 4, 7, 13, 59, 38, 550927, tzinfo=&lt;TimezoneInfo [UTC, GMT, +00:00:00, STD]&gt;), 'start_date': None, 'end_date': None, 'duration': None, 'state': None, 'try_number': 0, 'max_tries': 1, 'hostname': '', 'unixname': 'root', 'job_id': None, 'pool': 'default_pool', 'queue': 'default', 'priority_weight': 2, 'operator': None, 'queued_dttm': None, 'pid': None, 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fd6534586f0&gt;}  ... displaying 10 of 7120 total bound parameter sets ...  {'task_id': 'train_model_ZYYJ_ein', 'dag_id': 'predict_model_training_combined-2', 'execution_date': datetime.datetime(2020, 4, 7, 13, 59, 38, 550927, tzinfo=&lt;TimezoneInfo [UTC, GMT, +00:00:00, STD]&gt;), 'start_date': None, 'end_date': None, 'duration': None, 'state': None, 'try_number': 0, 'max_tries': 1, 'hostname': '', 'unixname': 'root', 'job_id': None, 'pool': 'default_pool', 'queue': 'default', 'priority_weight': 3, 'operator': None, 'queued_dttm': None, 'pid': None, 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fd652758660&gt;}, {'task_id': 'watch_training_ZYYJ_ein', 'dag_id': 'predict_model_training_combined-2', 'execution_date': datetime.datetime(2020, 4, 7, 13, 59, 38, 550927, tzinfo=&lt;TimezoneInfo [UTC, GMT, +00:00:00, STD]&gt;), 'start_date': None, 'end_date': None, 'duration': None, 'state': None, 'try_number': 0, 'max_tries': 1, 'hostname': '', 'unixname': 'root', 'job_id': None, 'pool': 'default_pool', 'queue': 'default', 'priority_weight': 2, 'operator': None, 'queued_dttm': None, 'pid': None, 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fd6527586f0&gt;})]
(Background on this error at: http://sqlalche.me/e/gkpj)
Process DagFileProcessor104982-Process:
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 1226, in _execute_context
    cursor, statement, parameters, context
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/dialects/postgresql/psycopg2.py", line 854, in do_executemany
    cursor.executemany(statement, parameters)
psycopg2.errors.UniqueViolation: duplicate key value violates unique constraint "task_instance_pkey"
DETAIL:  Key (task_id, dag_id, execution_date)=(create_cluster, predict_model_training_combined-4, 2020-04-07 14:41:02.111727+00) already exists.
</code></pre></div>