Python Airflow DAG无法分析{{ds}}

发布于01月26日

我有一个气流DAG，我使用{{ds}}来获得合理的日期.根据Airflow文档模板{{ds}}以字符串格式以YYYY-MM-DD格式返回逻辑日期.因此，我使用以下代码来处理日期

(datetime.strptime('{{ dag_run.logical_date|ds }}', '%Y-%m-%d') - timedelta(3)).strftime('%Y-%m-%d')

但是得到以下错误


Broken DAG: [/usr/local/airflow/dags/custom_dags/feature_store_daily.py] Traceback (most recent call last):
  File "/usr/lib/python3.10/_strptime.py", line 568, in _strptime_datetime
    tt, fraction, gmtoff_fraction = _strptime(data_string, format)
  File "/usr/lib/python3.10/_strptime.py", line 349, in _strptime
    raise ValueError("time data %r does not match format %r" %
ValueError: time data '{{dag_run.logical_date|ds}}' does not match format '%Y-%m-%d'

我不能弄清楚为什么我会收到这个错误.

请指教

推荐答案

首先，我认为问题可能是在"|"操作数前后没有足够的空间，但我已经测试过了，并且运行良好.

在这种情况下，主要问题是您没有在有效解析的地方使用JJJA模板.因此，错误在于您没有要格式化的日期，而是一个原封不动的字符串.

task_xx = SomeOperator(
    ...
    a="{{ dag_run.logical_date|ds }}",
    b="{{dag_run.logical_date | ds}}",
    c="{{ dag_run.logical_date | ds }}"
)

然后，在SomeOperator中，这些属性必须为template_fields，这样才能对其进行解析:

class SomeOperator(BaseOperator):
    template_fields = ['a', 'b', 'c']

   ....

现在，当你使用这些属性时，你会得到你想要的. 我已经测试过了，工作正常.

或者，您也可以将JJJA模板与已经接受它的字段一起使用.

Python Airflow DAG无法分析{{ds}}

推荐答案

Python相关问答推荐

将numpy矩阵映射到字符串矩阵

阅读Polars Python中管道的函数定义

opencv Python稳定的图标识别

拆分pandas列并创建包含这些拆分值计数的新列

GL pygame无法让缓冲区与vertextPointer和colorPointer一起可靠地工作

追溯(最近最后一次调用)：文件C：\Users\Diplom/PycharmProject\Yolo01\Roboflow-4.py，第4行，在模块导入roboflow中

查找两极rame中组之间的所有差异

按列分区，按另一列排序

运行终端命令时出现问题：pip start anonymous"

有没有一种方法可以从python的pussompy比较结果中提取文本？

如何让这个星型模式在Python中只使用一个for循环？

移动条情节旁边的半小提琴情节在海运

如何在turtle中不使用write()来绘制填充字母(例如OEG)

为什么Django管理页面和我的页面的其他CSS文件和图片都找不到？'

可以bcrypts AES—256 GCM加密损坏ZIP文件吗？

如何使用Numpy. stracards重新编写滚动和？

Geopandas未返回正确的缓冲区(单位：米)

LocaleError：模块keras._' tf_keras. keras没有属性__internal_'''

从列表中获取n个元素，其中list [i][0]== value''

极柱内丢失类型信息""