Могут ли некоторые предоставить мне схему для воссоздания таблицы dag_run в airflow-db?

Question

У меня есть среда google cloud composer на GCP, и я случайно удалил таблицу dag_runs, из-за которой airflow_scheduler продолжал падать, а веб-сервер airflow не запускался.Мне удалось воссоздать таблицу dag_run в airflow-db, которая остановила сбой, но я думаю, что не правильно понял схему, так как получаю следующую ошибку, когда вручную запускаю dag на веб-сервере airflow.

Ooops.

                      ____/ (  (    )   )  \___
                     /( (  (  )   _    ))  )   )\
                   ((     (   )(    )  )   (   )  )
                 ((/  ( _(   )   (   _) ) (  () )  )
                ( (  ( (_)   ((    (   )  .((_ ) .  )_
               ( (  )    (      (  )    )   ) . ) (   )
              (  (   (  (   ) (  _  ( _) ).  ) . ) ) ( )
              ( (  (   ) (  )   (  ))     ) _)(   )  )  )
             ( (  ( \ ) (    (_  ( ) ( )  )   ) )  )) ( )
              (  (   (  (   (_ ( ) ( _    )  ) (  )  )   )
             ( (  ( (  (  )     (_  )  ) )  _)   ) _( ( )
              ((  (   )(    (     _    )   _) _(_ (  (_ )
               (_((__(_(__(( ( ( |  ) ) ) )_))__))_)___)
               ((__)        \\||lll|l||///          \_))
                        (   /(/ (  )  ) )\   )
                      (    ( ( ( | | ) ) )\   )
                       (   /(| / ( )) ) ) )) )
                     (     ( ((((_(|)_)))))     )
                      (      ||\(|(|)|/||     )
                    (        |(||(||)||||        )
                      (     //|/l|||)|\\ \     )

(/ / // / | // |||| \ \ \ \ _)

Узел: 38b47b3e06a1

Traceback (most recent call last):
  File "/opt/python3.6/lib/python3.6/site-packages/flask/app.py", line 1988, in wsgi_app
    response = self.full_dispatch_request()
  File "/opt/python3.6/lib/python3.6/site-packages/flask/app.py", line 1641, in full_dispatch_request
    rv = self.handle_user_exception(e)
  File "/opt/python3.6/lib/python3.6/site-packages/flask/app.py", line 1544, in handle_user_exception
    reraise(exc_type, exc_value, tb)
  File "/opt/python3.6/lib/python3.6/site-packages/flask/_compat.py", line 33, in reraise
    raise value
  File "/opt/python3.6/lib/python3.6/site-packages/flask/app.py", line 1639, in full_dispatch_request
    rv = self.dispatch_request()
  File "/opt/python3.6/lib/python3.6/site-packages/flask/app.py", line 1625, in dispatch_request
    return self.view_functions[rule.endpoint](**req.view_args)
  File "/opt/python3.6/lib/python3.6/site-packages/flask_admin/base.py", line 69, in inner
    return self._run_view(f, *args, **kwargs)
  File "/opt/python3.6/lib/python3.6/site-packages/flask_admin/base.py", line 368, in _run_view
    return fn(self, *args, **kwargs)
  File "/opt/python3.6/lib/python3.6/site-packages/flask_login.py", line 755, in decorated_view
    return func(*args, **kwargs)
  File "/usr/local/lib/airflow/airflow/www/utils.py", line 262, in wrapper
    return f(*args, **kwargs)
  File "/usr/local/lib/airflow/airflow/www/utils.py", line 309, in wrapper
    return f(*args, **kwargs)
  File "/usr/local/lib/airflow/airflow/www/views.py", line 929, in trigger
    external_trigger=True
  File "/usr/local/lib/airflow/airflow/utils/db.py", line 50, in wrapper
    result = func(*args, **kwargs)
  File "/usr/local/lib/airflow/airflow/models.py", line 3781, in create_dagrun
    run.refresh_from_db()
  File "/usr/local/lib/airflow/airflow/utils/db.py", line 50, in wrapper
    result = func(*args, **kwargs)
  File "/usr/local/lib/airflow/airflow/models.py", line 4439, in refresh_from_db
    DR.run_id == self.run_id
  File "/opt/python3.6/lib/python3.6/site-packages/sqlalchemy/orm/query.py", line 3077, in one
    raise orm_exc.NoResultFound("No row was found for one()")
sqlalchemy.orm.exc.NoResultFound: No row was found for one()

Answer 1

DagRun SQLAlchemy модель могла бы помочь немного

id = Column(Integer, primary_key=True)
dag_id = Column(String(ID_LEN))
execution_date = Column(UtcDateTime, default=timezone.utcnow)
start_date = Column(UtcDateTime, default=timezone.utcnow)
end_date = Column(UtcDateTime)
_state = Column('state', String(50), default=State.RUNNING)
run_id = Column(String(ID_LEN))
external_trigger = Column(Boolean, default=True)
conf = Column(PickleType)

но, тем не менее, вот MySQL инструкция DDL

mysql> SHOW CREATE TABLE `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=177
DEFAULT CHARSET=utf8mb4
COLLATE=utf8mb4_0900_ai_ci

и описание таблицы

mysql> DESC dag_run;
+------------------+--------------+------+-----+---------+----------------+
| Field            | Type         | Null | Key | Default | Extra          |
+------------------+--------------+------+-----+---------+----------------+
| id               | int(11)      | NO   | PRI | NULL    | auto_increment |
| dag_id           | varchar(250) | YES  | MUL | NULL    |                |
| execution_date   | timestamp(6) | YES  |     | NULL    |                |
| state            | varchar(50)  | YES  |     | NULL    |                |
| run_id           | varchar(250) | YES  |     | NULL    |                |
| external_trigger | tinyint(1)   | YES  |     | NULL    |                |
| conf             | blob         | YES  |     | NULL    |                |
| end_date         | timestamp(6) | YES  |     | NULL    |                |
| start_date       | timestamp(6) | YES  |     | NULL    |                |
+------------------+--------------+------+-----+---------+----------------+