Trigger DAG will get (_mysql_exceptions.IntegrityError) Duplicate entry 'xxxxx' for key 'dag_id'
Opened this issue · 1 comments
i have a dag: "pcdn_export_agg_peak_daily", it will be trigger dag "pcdn_export_agg_peak" 3 times by difference run_id
and dag_run.conf
。
but when dag: "pcdn_export_agg_peak_daily" is running end, log show it get erros and exit with code 1:
[2019-11-21 17:47:56,825] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx sqlalchemy.exc.IntegrityError: (_mysql_exceptions.IntegrityError) (1062, "Duplicate entry 'pcdn_export_agg_peak.split_to_agg_9.pcdn_agg-2019-11-21 09:47:00' for key 'dag_id'")
[2019-11-21 17:47:56,825] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx [SQL: INSERT INTO dag_run (dag_id, execution_date, start_date, end_date, state, run_id, external_trigger, conf) VALUES (%s, %s, %s, %s, %s, %s, %s, %s)]
[2019-11-21 17:47:56,825] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx [parameters: ('pcdn_export_agg_peak.split_to_agg_9.pcdn_agg', <Pendulum [2019-11-21T09:47:00+00:00]>, datetime.datetime(2019, 11, 21, 9, 47, 56, 409081, tzinfo=<Timezone [UTC]>), None, 'running', 'tri_peak_agg-daily_device_app_tx-for:2019-11-20-on:20191120013000.000000', 1, b'\x80\x04\x95&\x01\x00\x00\x00\x00\x00\x00}\x94(\x8c\x03env\x94\x8c\x03dev\x94\x8c\x08start_ts\x94J\x80=\xd5]\x8c\x06end_ts\x94J\xa4K\xd5]\x8c\tstat_ ... (275 characters truncated) ... \x8c\x06device\x94as\x8c\tlog_level\x94\x8c\x04INFO\x94\x8c\rseries_chunks\x94Kd\x8c\tsp_chunks\x94J@B\x0f\x00\x8c\nsp_schunks\x94J\xa0\x86\x01\x00u.')]
[2019-11-21 17:47:56,825] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx (Background on this error at: http://sqlalche.me/e/gkpj)
The error message indicates that there is a duplicate key: "pcdn_export_agg_peak.split_to_agg_9.pcdn_agg-2019-11-21 09:47:00", because I am preparing to insert the data is including these fields: ("dag_id", "execution_date") VALUES is ("pcdn_export_agg_peak.split_to_agg_9.pcdn_agg", <Pendulum [2019-11-21T09:47:00+00:00]>).
my airflow version is v1.10.5
, backend meatadata database is mysql.
in the mysql database airflow
, show create table dag_run
; result is this:
----------------------------------------------------------------+
| dag_run | CREATE TABLE `dag_run` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`dag_id` varchar(250) DEFAULT NULL,
`execution_date` timestamp(6) NULL DEFAULT NULL,
`state` varchar(50) DEFAULT NULL,
`run_id` varchar(250) DEFAULT NULL,
`external_trigger` tinyint(1) DEFAULT NULL,
`conf` blob,
`end_date` timestamp(6) NULL DEFAULT NULL,
`start_date` timestamp(6) NULL DEFAULT NULL,
PRIMARY KEY (`id`),
UNIQUE KEY `dag_id` (`dag_id`,`execution_date`),
UNIQUE KEY `dag_id_2` (`dag_id`,`run_id`),
KEY `dag_id_state` (`dag_id`,`state`)
) ENGINE=InnoDB AUTO_INCREMENT=27 DEFAULT CHARSET=latin1 |
this table has UNIQUE KEY dag_id
(dag_id
,execution_date
), this value is not repeatable.
However, I queried the value of "2019-11-21 09:47:00" to the data of "execution_date" and did not query it.
However, the value of "execution_date" of the data I inserted is "UTC 2019-11-21T09:47:00+00:00". If I convert according to my local time, the value should be "2019-11-21 17:47:00". I queried the value of "2019-11-21 17:47:00" to the data and get 4 result:
mysql> select `id`,`dag_id`,`execution_date`,`state`,`run_id`,`external_trigger`,`end_date`,`start_date` from `dag_run` where `execution_date`='2019-11-21 17:47:00';
+----+--------------------------------------------------+----------------------------+---------+--------------------------------------------------------------------------+------------------+----------+----------------------------+
| id | dag_id | execution_date | state | run_id | external_trigger | end_date | start_date |
+----+--------------------------------------------------+----------------------------+---------+--------------------------------------------------------------------------+------------------+----------+----------------------------+
| 8 | pcdn_export_agg_peak | 2019-11-21 17:47:00.000000 | running | tri_peak_agg-daily_device_app_tx-for:2019-11-20-on:20191120013000.000000 | 1 | NULL | 2019-11-21 17:47:53.924978 |
| 13 | pcdn_export_agg_peak.split_to_agg_9.pcdn_agg | 2019-11-21 17:47:00.000000 | running | tri_peak_agg-daily_device_app_tx-for:2019-11-20-on:20191120013000.000000 | 1 | NULL | 2019-11-21 17:47:54.530647 |
| 18 | pcdn_export_agg_peak.split_to_agg_9.partial_sort | 2019-11-21 17:47:00.000000 | running | tri_peak_agg-daily_device_app_tx-for:2019-11-20-on:20191120013000.000000 | 1 | NULL | 2019-11-21 17:47:55.154685 |
| 23 | pcdn_export_agg_peak.split_to_agg_9 | 2019-11-21 17:47:00.000000 | running | tri_peak_agg-daily_device_app_tx-for:2019-11-20-on:20191120013000.000000 | 1 | NULL | 2019-11-21 17:47:55.788638 |
+----+--------------------------------------------------+----------------------------+---------+--------------------------------------------------------------------------+------------------+----------+----------------------------+
4 rows in set (0.05 sec)
so, by the log error information, i guess the bug how cause of the failure.
-
airflow query dag_run by UNIQUE KEY
pcdn_export_agg_peak.split_to_agg_9.pcdn_agg-2019-11-21 09:47:00
and did not query data. -
and airflow insert data in to dag_run again, and trigger this bug..
For reference....
Complete error log:
*** Reading local file: /home/work/airflow/logs/pcdn_export_agg_peak_daily/peak_agg.daily_device_app_tx/2019-11-20T01:30:00+00:00/1.log
[2019-11-21 17:47:42,165] {taskinstance.py:620} INFO - Dependencies all met for <TaskInstance: pcdn_export_agg_peak_daily.peak_agg.daily_device_app_tx 2019-11-20T01:30:00+00:00 [queued]>
[2019-11-21 17:47:42,251] {taskinstance.py:620} INFO - Dependencies all met for <TaskInstance: pcdn_export_agg_peak_daily.peak_agg.daily_device_app_tx 2019-11-20T01:30:00+00:00 [queued]>
[2019-11-21 17:47:42,251] {taskinstance.py:838} INFO -
--------------------------------------------------------------------------------
[2019-11-21 17:47:42,251] {taskinstance.py:839} INFO - Starting attempt 1 of 1
[2019-11-21 17:47:42,251] {taskinstance.py:840} INFO -
--------------------------------------------------------------------------------
[2019-11-21 17:47:42,475] {taskinstance.py:859} INFO - Executing <Task(TriggerDagRunOperator): peak_agg.daily_device_app_tx> on 2019-11-20T01:30:00+00:00
[2019-11-21 17:47:42,475] {base_task_runner.py:133} INFO - Running: ['airflow', 'run', 'pcdn_export_agg_peak_daily', 'peak_agg.daily_device_app_tx', '2019-11-20T01:30:00+00:00', '--job_id', '2', '--pool', 'default_pool', '--raw', '-sd', 'DAGS_FOLDER/pcdn_export_agg_peak_daily.py', '--cfg_path', '/tmp/tmpvw0s95my']
[2019-11-21 17:47:43,148] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx [2019-11-21 17:47:43,148] {settings.py:213} INFO - settings.configure_orm(): Using pool settings. pool_size=10, max_overflow=20, pool_recycle=1800, pid=7815
[2019-11-21 17:47:43,460] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx [2019-11-21 17:47:43,458] {__init__.py:51} INFO - Using executor CeleryExecutor
[2019-11-21 17:47:44,505] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx [2019-11-21 17:47:44,504] {dagbag.py:90} INFO - Filling up the DagBag from /home/work/pcdn/offline_agg/dags/pcdn_export_agg_peak_daily.py
[2019-11-21 17:47:45,367] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx /home/work/pcdn/src/lib/conf_manager.py:21: YAMLLoadWarning: calling yaml.load() without Loader=... is deprecated, as the default Loader is unsafe. Please read https://msg.pyyaml.org/load for full details.
[2019-11-21 17:47:45,367] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx self._main_cf = yaml.load(fstream)
[2019-11-21 17:47:45,389] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx /home/work/pcdn/src/lib/conf_manager.py:24: YAMLLoadWarning: calling yaml.load() without Loader=... is deprecated, as the default Loader is unsafe. Please read https://msg.pyyaml.org/load for full details.
[2019-11-21 17:47:45,389] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx self._base_cf = yaml.load(bstream) or {}
[2019-11-21 17:47:46,885] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx [2019-11-21 17:47:46,884] {cli.py:516} INFO - Running <TaskInstance: pcdn_export_agg_peak_daily.peak_agg.daily_device_app_tx 2019-11-20T01:30:00+00:00 [running]> on host jh02-db2.jh02
[2019-11-21 17:47:47,293] {logging_mixin.py:95} INFO - context: {'dag': <DAG: pcdn_export_agg_peak_daily>, 'ds': '2019-11-20', 'next_ds': '2019-11-21', 'next_ds_nodash': '20191121', 'prev_ds': '2019-11-19', 'prev_ds_nodash': '20191119', 'ds_nodash': '20191120', 'ts': '2019-11-20T01:30:00+00:00', 'ts_nodash': '20191120T013000', 'ts_nodash_with_tz': '20191120T013000+0000', 'yesterday_ds': '2019-11-19', 'yesterday_ds_nodash': '20191119', 'tomorrow_ds': '2019-11-21', 'tomorrow_ds_nodash': '20191121', 'END_DATE': '2019-11-20', 'end_date': '2019-11-20', 'dag_run': <DagRun pcdn_export_agg_peak_daily @ 2019-11-20 01:30:00+00:00: scheduled__2019-11-20T01:30:00+00:00, externally triggered: False>, 'run_id': 'scheduled__2019-11-20T01:30:00+00:00', 'execution_date': <Pendulum [2019-11-20T01:30:00+00:00]>, 'prev_execution_date': <Pendulum [2019-11-19T01:30:00+00:00]>, 'prev_execution_date_success': <Proxy at 0x7fda2a4d7548 with factory <function TaskInstance.get_template_context.<locals>.<lambda> at 0x7fda3f7d02f0>>, 'prev_start_date_success': <Proxy at 0x7fda2a4d7508 with factory <function TaskInstance.get_template_context.<locals>.<lambda> at 0x7fda2a434d90>>, 'next_execution_date': <Pendulum [2019-11-21T01:30:00+00:00]>, 'latest_date': '2019-11-20', 'macros': <module 'airflow.macros' from '/home/work/pcdn/venv/lib/python3.6/site-packages/airflow/macros/__init__.py'>, 'params': {'stat_cfg': {'env': 'dev', 'start_ts': 1574256000, 'end_ts': 1574259620, 'stat_name': 'daily_device_app_tx', 'exporter': {'database': 'aegis', 'retention': 'one_week', 'measurements': 'docker_container', 'fields': 'tx,rx'}, 'order_by': {'tags': ['device']}, 'log_level': 'INFO', 'series_chunks': 100, 'sp_chunks': 1000000, 'sp_schunks': 100000}}, 'tables': None, 'task': <Task(TriggerDagRunOperator): peak_agg.daily_device_app_tx>, 'task_instance': <TaskInstance: pcdn_export_agg_peak_daily.peak_agg.daily_device_app_tx 2019-11-20T01:30:00+00:00 [running]>, 'ti': <TaskInstance: pcdn_export_agg_peak_daily.peak_agg.daily_device_app_tx 2019-11-20T01:30:00+00:00 [running]>, 'task_instance_key_str': 'pcdn_export_agg_peak_daily__peak_agg.daily_device_app_tx__20191120', 'conf': <module 'airflow.configuration' from '/home/work/pcdn/venv/lib/python3.6/site-packages/airflow/configuration.py'>, 'test_mode': False, 'var': {'value': None, 'json': None}, 'inlets': [], 'outlets': []}
[2019-11-21 17:47:47,293] {logging_mixin.py:95} INFO - dag config: {'env': 'dev', 'start_ts': 1574256000, 'end_ts': 1574259620, 'stat_name': 'daily_device_app_tx', 'exporter': {'database': 'aegis', 'retention': 'one_week', 'measurements': 'docker_container', 'fields': 'tx,rx'}, 'order_by': {'tags': ['device']}, 'log_level': 'INFO', 'series_chunks': 100, 'sp_chunks': 1000000, 'sp_schunks': 100000}
[2019-11-21 17:47:47,293] {logging_mixin.py:95} INFO - run_dag_id: tri_peak_agg-daily_device_app_tx-for:2019-11-20-on:20191120013000.000000
[2019-11-21 17:47:47,498] {logging_mixin.py:95} INFO - [�[34m2019-11-21 17:47:47,498�[0m] {�[34mdagbag.py:�[0m90} INFO�[0m - Filling up the DagBag from �[1m/home/work/pcdn/offline_agg/dags/pcdn_export_agg_peak_once.py�[0m�[0m
[2019-11-21 17:47:56,478] {taskinstance.py:1051} ERROR - (_mysql_exceptions.IntegrityError) (1062, "Duplicate entry 'pcdn_export_agg_peak.split_to_agg_9.pcdn_agg-2019-11-21 09:47:00' for key 'dag_id'")
[SQL: INSERT INTO dag_run (dag_id, execution_date, start_date, end_date, state, run_id, external_trigger, conf) VALUES (%s, %s, %s, %s, %s, %s, %s, %s)]
[parameters: ('pcdn_export_agg_peak.split_to_agg_9.pcdn_agg', <Pendulum [2019-11-21T09:47:00+00:00]>, datetime.datetime(2019, 11, 21, 9, 47, 56, 409081, tzinfo=<Timezone [UTC]>), None, 'running', 'tri_peak_agg-daily_device_app_tx-for:2019-11-20-on:20191120013000.000000', 1, b'\x80\x04\x95&\x01\x00\x00\x00\x00\x00\x00}\x94(\x8c\x03env\x94\x8c\x03dev\x94\x8c\x08start_ts\x94J\x80=\xd5]\x8c\x06end_ts\x94J\xa4K\xd5]\x8c\tstat_ ... (275 characters truncated) ... \x8c\x06device\x94as\x8c\tlog_level\x94\x8c\x04INFO\x94\x8c\rseries_chunks\x94Kd\x8c\tsp_chunks\x94J@B\x0f\x00\x8c\nsp_schunks\x94J\xa0\x86\x01\x00u.')]
(Background on this error at: http://sqlalche.me/e/gkpj)
Traceback (most recent call last):
File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1244, in _execute_context
cursor, statement, parameters, context
File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/engine/default.py", line 552, in do_execute
cursor.execute(statement, parameters)
File "/home/work/pcdn/venv/lib/python3.6/site-packages/MySQLdb/cursors.py", line 255, in execute
self.errorhandler(self, exc, value)
File "/home/work/pcdn/venv/lib/python3.6/site-packages/MySQLdb/connections.py", line 50, in defaulterrorhandler
raise errorvalue
File "/home/work/pcdn/venv/lib/python3.6/site-packages/MySQLdb/cursors.py", line 252, in execute
res = self._query(query)
File "/home/work/pcdn/venv/lib/python3.6/site-packages/MySQLdb/cursors.py", line 378, in _query
db.query(q)
File "/home/work/pcdn/venv/lib/python3.6/site-packages/MySQLdb/connections.py", line 280, in query
_mysql.connection.query(self, query)
_mysql_exceptions.IntegrityError: (1062, "Duplicate entry 'pcdn_export_agg_peak.split_to_agg_9.pcdn_agg-2019-11-21 09:47:00' for key 'dag_id'")
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "/home/work/pcdn/venv/lib/python3.6/site-packages/airflow/models/taskinstance.py", line 926, in _run_raw_task
result = task_copy.execute(context=context)
File "/home/work/pcdn/venv/lib/python3.6/site-packages/airflow/operators/dagrun_operator.py", line 95, in execute
replace_microseconds=False)
File "/home/work/pcdn/venv/lib/python3.6/site-packages/airflow/api/common/experimental/trigger_dag.py", line 126, in trigger_dag
replace_microseconds=replace_microseconds,
File "/home/work/pcdn/venv/lib/python3.6/site-packages/airflow/api/common/experimental/trigger_dag.py", line 90, in _trigger_dag
external_trigger=True,
File "/home/work/pcdn/venv/lib/python3.6/site-packages/airflow/utils/db.py", line 74, in wrapper
return func(*args, **kwargs)
File "/home/work/pcdn/venv/lib/python3.6/site-packages/airflow/models/dag.py", line 1286, in create_dagrun
session.commit()
File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 1027, in commit
self.transaction.commit()
File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 494, in commit
self._prepare_impl()
File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 473, in _prepare_impl
self.session.flush()
File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 2459, in flush
self._flush(objects)
File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 2597, in _flush
transaction.rollback(_capture_exception=True)
File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/util/langhelpers.py", line 68, in __exit__
compat.reraise(exc_type, exc_value, exc_tb)
File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/util/compat.py", line 153, in reraise
raise value
File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 2557, in _flush
flush_context.execute()
File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/orm/unitofwork.py", line 422, in execute
rec.execute(self)
File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/orm/unitofwork.py", line 589, in execute
uow,
File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/orm/persistence.py", line 245, in save_obj
insert,
File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/orm/persistence.py", line 1138, in _emit_insert_statements
statement, params
File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 988, in execute
return meth(self, multiparams, params)
File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/sql/elements.py", line 287, in _execute_on_connection
return connection._execute_clauseelement(self, multiparams, params)
File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1107, in _execute_clauseelement
distilled_params,
File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1248, in _execute_context
e, statement, parameters, cursor, context
File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1466, in _handle_dbapi_exception
util.raise_from_cause(sqlalchemy_exception, exc_info)
File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/util/compat.py", line 398, in raise_from_cause
reraise(type(exception), exception, tb=exc_tb, cause=cause)
File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/util/compat.py", line 152, in reraise
raise value.with_traceback(tb)
File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1244, in _execute_context
cursor, statement, parameters, context
File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/engine/default.py", line 552, in do_execute
cursor.execute(statement, parameters)
File "/home/work/pcdn/venv/lib/python3.6/site-packages/MySQLdb/cursors.py", line 255, in execute
self.errorhandler(self, exc, value)
File "/home/work/pcdn/venv/lib/python3.6/site-packages/MySQLdb/connections.py", line 50, in defaulterrorhandler
raise errorvalue
File "/home/work/pcdn/venv/lib/python3.6/site-packages/MySQLdb/cursors.py", line 252, in execute
res = self._query(query)
File "/home/work/pcdn/venv/lib/python3.6/site-packages/MySQLdb/cursors.py", line 378, in _query
db.query(q)
File "/home/work/pcdn/venv/lib/python3.6/site-packages/MySQLdb/connections.py", line 280, in query
_mysql.connection.query(self, query)
sqlalchemy.exc.IntegrityError: (_mysql_exceptions.IntegrityError) (1062, "Duplicate entry 'pcdn_export_agg_peak.split_to_agg_9.pcdn_agg-2019-11-21 09:47:00' for key 'dag_id'")
[SQL: INSERT INTO dag_run (dag_id, execution_date, start_date, end_date, state, run_id, external_trigger, conf) VALUES (%s, %s, %s, %s, %s, %s, %s, %s)]
[parameters: ('pcdn_export_agg_peak.split_to_agg_9.pcdn_agg', <Pendulum [2019-11-21T09:47:00+00:00]>, datetime.datetime(2019, 11, 21, 9, 47, 56, 409081, tzinfo=<Timezone [UTC]>), None, 'running', 'tri_peak_agg-daily_device_app_tx-for:2019-11-20-on:20191120013000.000000', 1, b'\x80\x04\x95&\x01\x00\x00\x00\x00\x00\x00}\x94(\x8c\x03env\x94\x8c\x03dev\x94\x8c\x08start_ts\x94J\x80=\xd5]\x8c\x06end_ts\x94J\xa4K\xd5]\x8c\tstat_ ... (275 characters truncated) ... \x8c\x06device\x94as\x8c\tlog_level\x94\x8c\x04INFO\x94\x8c\rseries_chunks\x94Kd\x8c\tsp_chunks\x94J@B\x0f\x00\x8c\nsp_schunks\x94J\xa0\x86\x01\x00u.')]
(Background on this error at: http://sqlalche.me/e/gkpj)
[2019-11-21 17:47:56,485] {taskinstance.py:1082} INFO - Marking task as FAILED.
[2019-11-21 17:47:56,816] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx Traceback (most recent call last):
[2019-11-21 17:47:56,816] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1244, in _execute_context
[2019-11-21 17:47:56,816] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx cursor, statement, parameters, context
[2019-11-21 17:47:56,816] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/engine/default.py", line 552, in do_execute
[2019-11-21 17:47:56,817] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx cursor.execute(statement, parameters)
[2019-11-21 17:47:56,817] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/MySQLdb/cursors.py", line 255, in execute
[2019-11-21 17:47:56,817] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx self.errorhandler(self, exc, value)
[2019-11-21 17:47:56,817] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/MySQLdb/connections.py", line 50, in defaulterrorhandler
[2019-11-21 17:47:56,817] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx raise errorvalue
[2019-11-21 17:47:56,817] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/MySQLdb/cursors.py", line 252, in execute
[2019-11-21 17:47:56,817] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx res = self._query(query)
[2019-11-21 17:47:56,817] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/MySQLdb/cursors.py", line 378, in _query
[2019-11-21 17:47:56,818] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx db.query(q)
[2019-11-21 17:47:56,818] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/MySQLdb/connections.py", line 280, in query
[2019-11-21 17:47:56,818] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx _mysql.connection.query(self, query)
[2019-11-21 17:47:56,818] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx _mysql_exceptions.IntegrityError: (1062, "Duplicate entry 'pcdn_export_agg_peak.split_to_agg_9.pcdn_agg-2019-11-21 09:47:00' for key 'dag_id'")
[2019-11-21 17:47:56,818] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx
[2019-11-21 17:47:56,818] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx The above exception was the direct cause of the following exception:
[2019-11-21 17:47:56,818] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx
[2019-11-21 17:47:56,818] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx Traceback (most recent call last):
[2019-11-21 17:47:56,819] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/bin/airflow", line 32, in <module>
[2019-11-21 17:47:56,819] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx args.func(args)
[2019-11-21 17:47:56,819] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/airflow/utils/cli.py", line 74, in wrapper
[2019-11-21 17:47:56,819] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx return f(*args, **kwargs)
[2019-11-21 17:47:56,819] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/airflow/bin/cli.py", line 522, in run
[2019-11-21 17:47:56,819] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx _run(args, dag, ti)
[2019-11-21 17:47:56,819] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/airflow/bin/cli.py", line 440, in _run
[2019-11-21 17:47:56,819] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx pool=args.pool,
[2019-11-21 17:47:56,819] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/airflow/utils/db.py", line 74, in wrapper
[2019-11-21 17:47:56,820] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx return func(*args, **kwargs)
[2019-11-21 17:47:56,820] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/airflow/models/taskinstance.py", line 926, in _run_raw_task
[2019-11-21 17:47:56,820] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx result = task_copy.execute(context=context)
[2019-11-21 17:47:56,820] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/airflow/operators/dagrun_operator.py", line 95, in execute
[2019-11-21 17:47:56,820] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx replace_microseconds=False)
[2019-11-21 17:47:56,820] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/airflow/api/common/experimental/trigger_dag.py", line 126, in trigger_dag
[2019-11-21 17:47:56,820] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx replace_microseconds=replace_microseconds,
[2019-11-21 17:47:56,820] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/airflow/api/common/experimental/trigger_dag.py", line 90, in _trigger_dag
[2019-11-21 17:47:56,821] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx external_trigger=True,
[2019-11-21 17:47:56,821] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/airflow/utils/db.py", line 74, in wrapper
[2019-11-21 17:47:56,821] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx return func(*args, **kwargs)
[2019-11-21 17:47:56,821] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/airflow/models/dag.py", line 1286, in create_dagrun
[2019-11-21 17:47:56,821] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx session.commit()
[2019-11-21 17:47:56,821] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 1027, in commit
[2019-11-21 17:47:56,821] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx self.transaction.commit()
[2019-11-21 17:47:56,821] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 494, in commit
[2019-11-21 17:47:56,821] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx self._prepare_impl()
[2019-11-21 17:47:56,822] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 473, in _prepare_impl
[2019-11-21 17:47:56,822] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx self.session.flush()
[2019-11-21 17:47:56,822] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 2459, in flush
[2019-11-21 17:47:56,822] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx self._flush(objects)
[2019-11-21 17:47:56,822] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 2597, in _flush
[2019-11-21 17:47:56,822] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx transaction.rollback(_capture_exception=True)
[2019-11-21 17:47:56,822] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/util/langhelpers.py", line 68, in __exit__
[2019-11-21 17:47:56,822] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx compat.reraise(exc_type, exc_value, exc_tb)
[2019-11-21 17:47:56,822] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/util/compat.py", line 153, in reraise
[2019-11-21 17:47:56,822] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx raise value
[2019-11-21 17:47:56,823] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 2557, in _flush
[2019-11-21 17:47:56,823] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx flush_context.execute()
[2019-11-21 17:47:56,823] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/orm/unitofwork.py", line 422, in execute
[2019-11-21 17:47:56,823] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx rec.execute(self)
[2019-11-21 17:47:56,823] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/orm/unitofwork.py", line 589, in execute
[2019-11-21 17:47:56,823] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx uow,
[2019-11-21 17:47:56,823] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/orm/persistence.py", line 245, in save_obj
[2019-11-21 17:47:56,823] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx insert,
[2019-11-21 17:47:56,823] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/orm/persistence.py", line 1138, in _emit_insert_statements
[2019-11-21 17:47:56,823] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx statement, params
[2019-11-21 17:47:56,823] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 988, in execute
[2019-11-21 17:47:56,823] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx return meth(self, multiparams, params)
[2019-11-21 17:47:56,823] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/sql/elements.py", line 287, in _execute_on_connection
[2019-11-21 17:47:56,823] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx return connection._execute_clauseelement(self, multiparams, params)
[2019-11-21 17:47:56,823] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1107, in _execute_clauseelement
[2019-11-21 17:47:56,823] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx distilled_params,
[2019-11-21 17:47:56,824] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1248, in _execute_context
[2019-11-21 17:47:56,824] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx e, statement, parameters, cursor, context
[2019-11-21 17:47:56,824] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1466, in _handle_dbapi_exception
[2019-11-21 17:47:56,824] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx util.raise_from_cause(sqlalchemy_exception, exc_info)
[2019-11-21 17:47:56,824] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/util/compat.py", line 398, in raise_from_cause
[2019-11-21 17:47:56,824] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx reraise(type(exception), exception, tb=exc_tb, cause=cause)
[2019-11-21 17:47:56,824] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/util/compat.py", line 152, in reraise
[2019-11-21 17:47:56,824] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx raise value.with_traceback(tb)
[2019-11-21 17:47:56,824] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1244, in _execute_context
[2019-11-21 17:47:56,824] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx cursor, statement, parameters, context
[2019-11-21 17:47:56,824] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/sqlalchemy/engine/default.py", line 552, in do_execute
[2019-11-21 17:47:56,824] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx cursor.execute(statement, parameters)
[2019-11-21 17:47:56,824] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/MySQLdb/cursors.py", line 255, in execute
[2019-11-21 17:47:56,824] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx self.errorhandler(self, exc, value)
[2019-11-21 17:47:56,824] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/MySQLdb/connections.py", line 50, in defaulterrorhandler
[2019-11-21 17:47:56,825] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx raise errorvalue
[2019-11-21 17:47:56,825] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/MySQLdb/cursors.py", line 252, in execute
[2019-11-21 17:47:56,825] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx res = self._query(query)
[2019-11-21 17:47:56,825] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/MySQLdb/cursors.py", line 378, in _query
[2019-11-21 17:47:56,825] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx db.query(q)
[2019-11-21 17:47:56,825] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx File "/home/work/pcdn/venv/lib/python3.6/site-packages/MySQLdb/connections.py", line 280, in query
[2019-11-21 17:47:56,825] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx _mysql.connection.query(self, query)
[2019-11-21 17:47:56,825] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx sqlalchemy.exc.IntegrityError: (_mysql_exceptions.IntegrityError) (1062, "Duplicate entry 'pcdn_export_agg_peak.split_to_agg_9.pcdn_agg-2019-11-21 09:47:00' for key 'dag_id'")
[2019-11-21 17:47:56,825] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx [SQL: INSERT INTO dag_run (dag_id, execution_date, start_date, end_date, state, run_id, external_trigger, conf) VALUES (%s, %s, %s, %s, %s, %s, %s, %s)]
[2019-11-21 17:47:56,825] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx [parameters: ('pcdn_export_agg_peak.split_to_agg_9.pcdn_agg', <Pendulum [2019-11-21T09:47:00+00:00]>, datetime.datetime(2019, 11, 21, 9, 47, 56, 409081, tzinfo=<Timezone [UTC]>), None, 'running', 'tri_peak_agg-daily_device_app_tx-for:2019-11-20-on:20191120013000.000000', 1, b'\x80\x04\x95&\x01\x00\x00\x00\x00\x00\x00}\x94(\x8c\x03env\x94\x8c\x03dev\x94\x8c\x08start_ts\x94J\x80=\xd5]\x8c\x06end_ts\x94J\xa4K\xd5]\x8c\tstat_ ... (275 characters truncated) ... \x8c\x06device\x94as\x8c\tlog_level\x94\x8c\x04INFO\x94\x8c\rseries_chunks\x94Kd\x8c\tsp_chunks\x94J@B\x0f\x00\x8c\nsp_schunks\x94J\xa0\x86\x01\x00u.')]
[2019-11-21 17:47:56,825] {base_task_runner.py:115} INFO - Job 2: Subtask peak_agg.daily_device_app_tx (Background on this error at: http://sqlalche.me/e/gkpj)
[2019-11-21 17:47:57,393] {logging_mixin.py:95} INFO - [�[34m2019-11-21 17:47:57,392�[0m] {�[34mlocal_task_job.py:�[0m105} INFO�[0m - Task exited with return code 1�[0m
Hi @YouZhengChuan, any news on this issue? Have the same problem in composer 1.10.15-airflow1.10.12