如何使用通配符在气流中使用S3KeySensor搜索s3文件

Question

您好，我正在使用 S3KeySensor 查找在特定分区中创建的镶木地板文件。由于文件名是从 spark 生成的，例如 (part-00499-e91c1af8-4352-4de9*)，bucket_key 应该是什么？以下代码失败。

    ex:bucket_key=f"inbound/phix/empnf/datasetdate={var_ds_date}/*.parquet"

s3_data_filechk=S3KeySensor(
         task_id='s3_data_filechk',
         bucket_name=data_bucket_name,
         bucket_key=f"inbound/phix/empnf/datasetdate={var_ds_date}/*.parquet/",
         timeout=60 * 30, # timeout in 30 minutes 
         poke_interval=60 * 5 # (seconds); checking file in every five minutes
     )

Answer 1

S3KeySensor 有参数 wildcard_match：bucket_key 是否应该被解释为 Unix 通配符模式。您可以使用它来设置您想要的逻辑。

示例：

sensor=S3KeySensor(
         ...
         bucket_key="*.parquet",
         wildcard_match=True
     )

如何使用通配符在气流中使用S3KeySensor搜索s3文件

How to use wild card character to search for s3 file using S3KeySensor in airflow

amazon-s3

airflow

mwaa