Airflow Xcom : How to cast byte array for value into text or json text in SQL?

Question

我正在调查随着时间的推移，哪些数据处理作业在它们各自的使用中花费的时间更长(对于我们系统的安装，它已经运行了好几个月)。它处理的数据文件的大小在大小上有几个数量级的变化，所以我想标准化处理时间之间的比较，以及锁定在 XCOM 变量值内的有效负载中的记录数。

我想构建一个 SQL View ，我可以用它来关联处理持续时间(结束开始)、文件大小与执行日期，以查看处理在其生命周期中的稳定性。

在在线文档中，有一些关于将 Python 序列化为 JSON 的示例，但是，我们的 Airflow 元数据存储在 PostGres 中，我想创建一个 SQL View ，该 View 可以提供与处理 dags/tasks 相关的统计数据和相关元数据的指标处理嵌套在 XCOM 值中的自身。

现在有没有人如何将 XCOM 字节值转换为可在 PostGres SQL 中解析的内容？

Answer 1

我面临同样的问题。在挖掘 Airflow 源后，发现了这个:

https://github.com/apache/airflow/blob/2bea3d74952d0d68d90e8bbc307ac3dfe8fcf2ff/airflow/models/xcom.py#L221]

在数据库中设置 XCOM 变量时，它将序列化它。在 airflow.cfg有一个设置enable_xcom_pickling = True .

if conf.getboolean('core', 'enable_xcom_pickling'):
    return pickle.dumps(value)

看起来字节数组正在被腌制，然后被存储 .这很烦人，因为我认为没有办法直接从 postgres 解开字节数组。

您还可以设置另一个标志，称为 donot_pickle = False .还不确定这有什么作用 - 进一步研究