如何解决Airflow/SQLAlchemy 错误 - 在加载/刷新处理程序中加载上下文已更改
我正在尝试使用千里眼的 db-cleanup dag 来清除我们 xcom 表中的元数据,但是当我运行它时,我收到以下警告,在我手动停止作业之前打印了数千次以便不删除我们的 MysqL 实例:
SAWarning: Loading context for <BaseXCom at 0x7f26f789b370> has changed within a load/refresh handler,suggesting a row refresh operation took place. If this event handler is expected to be emitting row refresh operations within an existing load or refresh operation,set restore_load_context=True when establishing the listener to ensure the context remains unchanged when the event handler completes.
其他清理任务工作正常,但特别是 xcom 表我遇到了问题。我们有成百上千个活动的 dag,因此 xcom 表几乎每两秒钟就会被写入一次。我认为这就是导致此错误的原因,即数据在查询时不断变化。
我一直无法找到导致此问题的原因或如何解决此问题的任何示例。我尝试按照 SQLAlchemy docs 添加 "restore_load_context":True
行,但没有用。
{
"airflow_db_model": XCom,"age_check_column": XCom.execution_date,"keep_last": False,"keep_last_filters": None,"keep_last_group_by": None,"restore_load_context":True
},
....
def cleanup_function(**context):
logging.info("Retrieving max_execution_date from XCom")
max_date = context["ti"].xcom_pull(
task_ids=print_configuration.task_id,key="max_date"
)
max_date = dateutil.parser.parse(max_date) # stored as iso8601 str in xcom
airflow_db_model = context["params"].get("airflow_db_model")
state = context["params"].get("state")
age_check_column = context["params"].get("age_check_column")
keep_last = context["params"].get("keep_last")
keep_last_filters = context["params"].get("keep_last_filters")
keep_last_group_by = context["params"].get("keep_last_group_by")
restore_load_context = context["params"].get("restore_load_context")
为了不在这里粘贴太多代码,我在db-cleanup dag中使用了相同的代码。有没有人遇到过这种情况并找到解决方法?
我对 sqlalchemy 非常缺乏经验,完全不确定该将代码放在哪里或如何处理。
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。