访问修改的气流变量作为气流中S3传感器的自定义参数

Question

让我们看看我能不能解释一下这件事。在

看，我从一个客户那里收到的S3文件格式不好。例如，日期以短划线显示，例如“2017年7月10日”。在

因为我想访问它们以便能够下载它们，首先我有一个任务是气流中的S3传感器。看起来是这样的：

xxx = S3KeySensor(
    task_id='task_name',
    bucket_key=BUCKET_KEY,
    wildcard_match=True,
    params={'yesterday_ds_formatted': ????},
    provide_context=True,
    bucket_name=BUCKET_NAME,
    s3_conn_id=S3_CONN_ID,
    timeout=18 * 60 * 60,
    poke_interval=120,
    dag=dag)

在Airflow控制台的Variables部分，我有一个bucket_键和模板变量{{detaily{uds}formatted}。在

例如：“folder1/folder2/folder3/blablablabla”-{{参数昨天已格式化}}*.csv文件

我需要修改该模板变量或其他内容，以便它获取{{detaily{uds}}并将“-”替换为“uu”。在

我该怎么做？我没办法让它成功。。。 我试着在设置参数时调用一个自定义的python函数，但是我不能访问“ds”，即使使用kwargs也不行。可以这么说，我似乎不能事先访问模板变量。在

谢谢你！！在

Answer 1

如果我理解正确的话，您希望为bucket_key参数使用jinja模板，但是S3KeySensor不支持这一点。在

一种简单的方法是从S3KeySensor中将一个自定义传感器子类化，如下所示：

TemplatedS3KeySensor(S3KeySensor):
  template_fields = ('bucket_key',)