Airflow 任务执行 SQLAlchemy 操作后停止运行：解决故障指南

python

2024-03-24 21:15:50

Airflow 任务在执行 SQLAlchemy 操作后停止运行：故障排除指南

问题

Airflow 任务在执行 SQLAlchemy 操作后停止运行是一个常见的错误，它可能给开发人员带来头疼。幸运的是，这个问题通常可以通过解决一些关键的根源原因来解决。

原因及解决方案

1. 异步操作

SQLAlchemy 会话默认使用异步操作，这意味着数据库操作可以在后台执行。如果任务在等待异步操作完成之前结束，则会导致任务停止运行。

解决方案：

使用 SQLAlchemy 的 scoped_session 创建会话，它会强制同步操作。

2. 未提交的事务

如果任务在未提交事务的情况下结束，则数据库更改将不会持久化。这会导致任务停止运行，并且以后的尝试将失败。

解决方案：

确保在任务结束前提交所有事务。

3. 资源泄漏

如果任务创建了会话或引擎，但没有正确关闭它们，则会导致资源泄漏。这可能会使任务停止运行，或者导致性能问题。

解决方案：

使用 with 语句或 finally 块来确保会话和引擎在任务结束时正确关闭。

4. 数据库连接问题

如果任务无法连接到数据库，则会停止运行。这可能是由于数据库不可用、连接参数不正确或网络问题造成的。

解决方案：

检查数据库连接，确保其正常运行并使用正确的连接参数。

5. SQLAlchemy 版本冲突

不同的 SQLAlchemy 版本可能导致不兼容性，从而导致任务停止运行。

解决方案：

确保所有组件（Airflow、SQLAlchemy 和数据库驱动程序）使用兼容的版本。

代码示例

以下是一个修改后的代码示例，解决了 Airflow 任务在执行 SQLAlchemy 操作后停止运行的问题：

@task
def metrics_aggregation_task():
    try:
        logging.info("Starting metrics aggregation task")

        logging.info("Initializing engines")
        LANGFUSE_DATABASE_URL = os.getenv("LANGFUSE_DATABASE_URL")
        langfuse_engine = create_engine(LANGFUSE_DATABASE_URL)

        AIRFLOW_DATABASE_URL = os.getenv("AIRFLOW_DATABASE_URL")
        airflow_engine = create_engine(AIRFLOW_DATABASE_URL)

        AUTOMODELLING_DATABASE_URL = os.getenv("MODELLING_DATABASE_URL")
        automodelling_engine = create_engine(AUTOMODELLING_DATABASE_URL)

        # Use scoped_session to force synchronous operations
        session_automodelling = scoped_session(Session(bind=automodelling_engine))
        session_langfuse = scoped_session(Session(bind=langfuse_engine))
        session_airflow = scoped_session(Session(bind=airflow_engine))

        logging.info("Sessions created")
        product_id_and_run_id_query = """
                SELECT "productId", "id", "metadata"  FROM "Dags"
                """

        logging.info("Fetching product_ids_and_run_ids")
        product_ids_and_run_ids = session_automodelling.execute(
            product_id_and_run_id_query
        ).fetchall()

        logging.info(f"Fetched product_ids_and_run_ids : {product_ids_and_run_ids}")

        # Ensure transactions are committed before task ends
        session_automodelling.commit()
        session_langfuse.commit()
        session_airflow.commit()

    except Exception as e:
        logging.error("Error occurred in metrics aggregation task")
        logging.exception(e)
        raise e

    finally:
        # Close all sessions and engines
        session_automodelling.close()
        session_langfuse.close()
        session_airflow.close()
        automodelling_engine.dispose()
        langfuse_engine.dispose()
        airflow_engine.dispose()