is any solution to migrate past logs and data? #14952

uplsh580 · 2021-03-23T10:31:02Z

uplsh580
Mar 23, 2021

I was using Airflow 1.10.6 with MySQL.
To use Airflow 2.0 version, I set up Airflow 2.0.1 with PostgreSQL in the new instance and copied only the Dag files.
And I wanted to migrate log in the past(MySQL) to PostgreSQL.

So I tried the procedure below to testing migration airflow 2.0.1 + MySQL -> airflow 2.0.1 + PostgreSQL
before airflow 1.10.6 + MySQL -> airflow 2.0.1 + PostgreSQL.

airflow db init (create tables in PostgeSQL)
Export Existing MySQL data using csv files (21 of 30 airflow tables)

SELECT * FROM celery_taskmeta INTO OUTFILE '/var/lib/mysql-files/celery_taskmeta.csv' CHARACTER SET utf8  FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n';
SELECT * FROM celery_tasksetmeta INTO OUTFILE '/var/lib/mysql-files/celery_tasksetmeta.csv' CHARACTER SET utf8  FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n';
SELECT * FROM connection INTO OUTFILE '/var/lib/mysql-files/connection.csv' CHARACTER SET utf8  FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n';
SELECT * FROM dag INTO OUTFILE '/var/lib/mysql-files/dag.csv' CHARACTER SET utf8  FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n';
SELECT * FROM dag_code INTO OUTFILE '/var/lib/mysql-files/dag_code.csv' CHARACTER SET utf8  FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n';
SELECT * FROM dag_pickle INTO OUTFILE '/var/lib/mysql-files/dag_pickle.csv' CHARACTER SET utf8  FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n';
SELECT * FROM dag_run INTO OUTFILE '/var/lib/mysql-files/dag_run.csv' CHARACTER SET utf8  FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n';
SELECT * FROM dag_tag INTO OUTFILE '/var/lib/mysql-files/dag_tag.csv' CHARACTER SET utf8  FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n';
SELECT * FROM import_error INTO OUTFILE '/var/lib/mysql-files/import_error.csv' CHARACTER SET utf8  FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n';
SELECT * FROM job INTO OUTFILE '/var/lib/mysql-files/job.csv' CHARACTER SET utf8  FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n';
SELECT * FROM log INTO OUTFILE '/var/lib/mysql-files/log.csv' CHARACTER SET utf8  FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n';
SELECT * FROM rendered_task_instance_fields INTO OUTFILE '/var/lib/mysql-files/rendered_task_instance_fields.csv' CHARACTER SET utf8  FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n';
SELECT * FROM sensor_instance INTO OUTFILE '/var/lib/mysql-files/sensor_instance.csv' CHARACTER SET utf8  FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n';
SELECT * FROM serialized_dag INTO OUTFILE '/var/lib/mysql-files/serialized_dag.csv' CHARACTER SET utf8  FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n';
SELECT * FROM sla_miss INTO OUTFILE '/var/lib/mysql-files/sla_miss.csv' CHARACTER SET utf8  FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n';
SELECT * FROM slot_pool INTO OUTFILE '/var/lib/mysql-files/slot_pool.csv' CHARACTER SET utf8  FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n';
SELECT * FROM task_fail INTO OUTFILE '/var/lib/mysql-files/task_fail.csv' CHARACTER SET utf8  FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n';
SELECT * FROM task_instance INTO OUTFILE '/var/lib/mysql-files/task_instance.csv' CHARACTER SET utf8  FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n';
SELECT * FROM task_reschedule INTO OUTFILE '/var/lib/mysql-files/task_reschedule.csv' CHARACTER SET utf8  FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n';
SELECT * FROM variable INTO OUTFILE '/var/lib/mysql-files/variable.csv' CHARACTER SET utf8  FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n';
SELECT * FROM xcom INTO OUTFILE '/var/lib/mysql-files/xcom.csv' CHARACTER SET utf8  FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n';

import to PostgreSQL using csv files

TRUNCATE TABLE celery_taskmeta CASCADE; COPY celery_taskmeta FROM '/var/lib/mysql-files/celery_taskmeta.csv'  DELIMITER ',';
TRUNCATE TABLE celery_tasksetmeta CASCADE; COPY celery_tasksetmeta FROM '/var/lib/mysql-files/celery_tasksetmeta.csv'  DELIMITER ',';
TRUNCATE TABLE connection CASCADE; COPY connection FROM '/var/lib/mysql-files/connection.csv'  DELIMITER ',';
TRUNCATE TABLE dag CASCADE; COPY dag FROM '/var/lib/mysql-files/dag.csv'  DELIMITER ',';
TRUNCATE TABLE dag_code CASCADE; COPY dag_code FROM '/var/lib/mysql-files/dag_code.csv'  DELIMITER ',';
TRUNCATE TABLE dag_pickle CASCADE; COPY dag_pickle FROM '/var/lib/mysql-files/dag_pickle.csv'  DELIMITER ',';
TRUNCATE TABLE dag_run CASCADE; COPY dag_run FROM '/var/lib/mysql-files/dag_run.csv'  DELIMITER ',';
TRUNCATE TABLE dag_tag CASCADE; COPY dag_tag FROM '/var/lib/mysql-files/dag_tag.csv'  DELIMITER ',';
TRUNCATE TABLE import_error CASCADE; COPY import_error FROM '/var/lib/mysql-files/import_error.csv'  DELIMITER ',';
TRUNCATE TABLE job CASCADE; COPY job FROM '/var/lib/mysql-files/job.csv'  DELIMITER ',';
TRUNCATE TABLE log CASCADE; COPY log FROM '/var/lib/mysql-files/log.csv'  DELIMITER ',';
TRUNCATE TABLE rendered_task_instance_fields CASCADE; COPY rendered_task_instance_fields FROM '/var/lib/mysql-files/rendered_task_instance_fields.csv'  DELIMITER ',';
TRUNCATE TABLE sensor_instance CASCADE; COPY sensor_instance FROM '/var/lib/mysql-files/sensor_instance.csv'  DELIMITER ',';
TRUNCATE TABLE serialized_dag CASCADE; COPY serialized_dag FROM '/var/lib/mysql-files/serialized_dag.csv'  DELIMITER ',';
TRUNCATE TABLE sla_miss CASCADE; COPY sla_miss FROM '/var/lib/mysql-files/sla_miss.csv'  DELIMITER ',';
TRUNCATE TABLE slot_pool CASCADE; COPY slot_pool FROM '/var/lib/mysql-files/slot_pool.csv'  DELIMITER ',';
TRUNCATE TABLE task_fail CASCADE; COPY task_fail FROM '/var/lib/mysql-files/task_fail.csv'  DELIMITER ',';
TRUNCATE TABLE task_instance CASCADE; COPY task_instance FROM '/var/lib/mysql-files/task_instance.csv'  DELIMITER ',';
TRUNCATE TABLE task_reschedule CASCADE; COPY task_reschedule FROM '/var/lib/mysql-files/task_reschedule.csv'  DELIMITER ',';
TRUNCATE TABLE variable CASCADE; COPY variable FROM '/var/lib/mysql-files/variable.csv'  DELIMITER ',';
TRUNCATE TABLE xcom CASCADE; COPY xcom FROM '/var/lib/mysql-files/xcom.csv'  DELIMITER ',';

Copy Dag file
Run airflow

airflow users create  --role Admin --username admin --email admin --firstname admin --lastname admin --password admin
 
airflow celery worker -D
airflow scheduler -D
airflow webserver -p 8080 -D

The following error occurs:

Something bad has happened.
Please consider letting us know by creating a bug report using GitHub.

Python version: 3.6.8
Airflow version: 2.0.1
Node: np-airflow-beta001-lde-jp2v-dev.lineinfra-dev.com
-------------------------------------------------------------------------------
Traceback (most recent call last):
  File "/home1/www/airflow2/venv_air2/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1277, in _execute_context
    cursor, statement, parameters, context
  File "/home1/www/airflow2/venv_air2/lib/python3.6/site-packages/sqlalchemy/engine/default.py", line 608, in do_execute
    cursor.execute(statement, parameters)
psycopg2.errors.UniqueViolation: duplicate key value violates unique constraint "log_pkey"
DETAIL:  Key (id)=(5) already exists.


The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/home1/www/airflow2/venv_air2/lib/python3.6/site-packages/flask/app.py", line 2447, in wsgi_app
    response = self.full_dispatch_request()
  File "/home1/www/airflow2/venv_air2/lib/python3.6/site-packages/flask/app.py", line 1952, in full_dispatch_request
    rv = self.handle_user_exception(e)
  File "/home1/www/airflow2/venv_air2/lib/python3.6/site-packages/flask/app.py", line 1821, in handle_user_exception
    reraise(exc_type, exc_value, tb)
  File "/home1/www/airflow2/venv_air2/lib/python3.6/site-packages/flask/_compat.py", line 39, in reraise
    raise value
  File "/home1/www/airflow2/venv_air2/lib/python3.6/site-packages/flask/app.py", line 1950, in full_dispatch_request
    rv = self.dispatch_request()
  File "/home1/www/airflow2/venv_air2/lib/python3.6/site-packages/flask/app.py", line 1936, in dispatch_request
    return self.view_functions[rule.endpoint](**req.view_args)
  File "/home1/www/airflow2/venv_air2/lib/python3.6/site-packages/airflow/www/auth.py", line 34, in decorated
    return func(*args, **kwargs)
  File "/home1/www/airflow2/venv_air2/lib/python3.6/site-packages/airflow/www/decorators.py", line 97, in view_func
    return f(*args, **kwargs)
  File "/home1/www/airflow2/venv_air2/lib/python3.6/site-packages/airflow/www/decorators.py", line 58, in wrapper
    session.add(log)
  File "/home1/www/.pyenv/versions/3.6.8/lib/python3.6/contextlib.py", line 88, in __exit__
    next(self.gen)
  File "/home1/www/airflow2/venv_air2/lib/python3.6/site-packages/airflow/utils/session.py", line 32, in create_session
    session.commit()
  File "/home1/www/airflow2/venv_air2/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 1046, in commit
    self.transaction.commit()
  File "/home1/www/airflow2/venv_air2/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 504, in commit
    self._prepare_impl()
  File "/home1/www/airflow2/venv_air2/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 483, in _prepare_impl
    self.session.flush()
  File "/home1/www/airflow2/venv_air2/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 2540, in flush
    self._flush(objects)
  File "/home1/www/airflow2/venv_air2/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 2682, in _flush
    transaction.rollback(_capture_exception=True)
  File "/home1/www/airflow2/venv_air2/lib/python3.6/site-packages/sqlalchemy/util/langhelpers.py", line 70, in __exit__
    with_traceback=exc_tb,
  File "/home1/www/airflow2/venv_air2/lib/python3.6/site-packages/sqlalchemy/util/compat.py", line 182, in raise_
    raise exception
  File "/home1/www/airflow2/venv_air2/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 2642, in _flush
    flush_context.execute()
  File "/home1/www/airflow2/venv_air2/lib/python3.6/site-packages/sqlalchemy/orm/unitofwork.py", line 422, in execute
    rec.execute(self)
  File "/home1/www/airflow2/venv_air2/lib/python3.6/site-packages/sqlalchemy/orm/unitofwork.py", line 589, in execute
    uow,
  File "/home1/www/airflow2/venv_air2/lib/python3.6/site-packages/sqlalchemy/orm/persistence.py", line 245, in save_obj
    insert,
  File "/home1/www/airflow2/venv_air2/lib/python3.6/site-packages/sqlalchemy/orm/persistence.py", line 1136, in _emit_insert_statements
    statement, params
  File "/home1/www/airflow2/venv_air2/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1011, in execute
    return meth(self, multiparams, params)
  File "/home1/www/airflow2/venv_air2/lib/python3.6/site-packages/sqlalchemy/sql/elements.py", line 298, in _execute_on_connection
    return connection._execute_clauseelement(self, multiparams, params)
  File "/home1/www/airflow2/venv_air2/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1130, in _execute_clauseelement
    distilled_params,
  File "/home1/www/airflow2/venv_air2/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1317, in _execute_context
    e, statement, parameters, cursor, context
  File "/home1/www/airflow2/venv_air2/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1511, in _handle_dbapi_exception
    sqlalchemy_exception, with_traceback=exc_info[2], from_=e
  File "/home1/www/airflow2/venv_air2/lib/python3.6/site-packages/sqlalchemy/util/compat.py", line 182, in raise_
    raise exception
  File "/home1/www/airflow2/venv_air2/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1277, in _execute_context
    cursor, statement, parameters, context
  File "/home1/www/airflow2/venv_air2/lib/python3.6/site-packages/sqlalchemy/engine/default.py", line 608, in do_execute
    cursor.execute(statement, parameters)
sqlalchemy.exc.IntegrityError: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "log_pkey"
DETAIL:  Key (id)=(5) already exists.

[SQL: INSERT INTO log (dttm, dag_id, task_id, event, execution_date, owner, extra) VALUES (%(dttm)s, %(dag_id)s, %(task_id)s, %(event)s, %(execution_date)s, %(owner)s, %(extra)s) RETURNING log.id]
[parameters: {'dttm': datetime.datetime(2021, 3, 23, 10, 25, 32, 162916, tzinfo=Timezone('UTC')), 'dag_id': 'tutorial', 'task_id': None, 'event': 'tree', 'execution_date': None, 'owner': 'admin', 'extra': "[('dag_id', 'tutorial')]"}]
(Background on this error at: http://sqlalche.me/e/13/gkpj)

Is there a migration solution?

turbaszek · 2021-03-25T19:08:44Z

turbaszek
Mar 25, 2021
Collaborator

Have you considered using airflow upgrade check?
https://airflow.apache.org/docs/apache-airflow/stable/upgrade-check.html

1 reply

uplsh580 Mar 25, 2021
Author

Thank you for your comments.
I missed some explanation.
The tried procedure was airflow 2.0.1 + MySQL -> airflow 2.0.1 + PostgreSQL for testing migration
before airflow 1.10.6 + MySQL -> airflow 2.0.1 + PostgreSQL.

Navigation Menu

Repository navigation

is any solution to migrate past logs and data? #14952

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{editor}}'s edit

{{editor}}'s edit

Uh oh!

Replies: 1 comment 1 reply

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{title}}

Uh oh!

Select a reply

Uh oh!

is any solution to migrate past logs and data? #14952

Uh oh!

Uh oh!

uplsh580 Mar 23, 2021

Replies: 1 comment · 1 reply

Uh oh!

turbaszek Mar 25, 2021 Collaborator

Uh oh!

uplsh580 Mar 25, 2021 Author

uplsh580
Mar 23, 2021

Replies: 1 comment 1 reply

turbaszek
Mar 25, 2021
Collaborator

uplsh580 Mar 25, 2021
Author