循环气流变量问题 [英] loop over airflow variables issue question

问题描述

我很难遍历脚本中的气流变量

I am having hard time looping over an airflow variable in my script

所以我需要在存储桶中列出所有以字符串为前缀的文件.

so I have a requirement to list all files prefixed by string in a bucket.

接下来遍历它并进行一些操作.

next loop through it and do some operations.

我尝试使用xcomm和subdags，但是我不知道为什么，所以我想出了一种新方法.

I tried making use of xcomm and subdags but i couldn't figure it out so i came up with a new approach.

尽管其中涉及2个脚本

第一个脚本将气流变量设置为我生成的值

1 st scripts sets the airflow variable with a value i generate

下面是代码.

#!/usr/bin/env python


with DAG('Test_variable', default_args=default_args,
    schedule_interval=None
) as dag:
    GCS_File_list = GoogleCloudStorageListOperator(
                    task_id= 'list_Files',
                    bucket= 'bucketname',
                    prefix='aaa',
                    delimiter='.csv',
                    google_cloud_storage_conn_id='google_cloud_default'
                    #provide_context = True,
                    #dag = dag
                )
    def update_variable(**context):
        files_list = Variable.get('files_list')
        print(type(Variable.get('files_list')))
        updated_file_list = context['ti'].xcom_pull(task_ids='list_Files')
        #updated_file_list = updated_file_list.strip('][').split(',')
        Variable.set("files_list", updated_file_list)
        #print(updated_file_list)
        #print(type(updated_file_list))

    python_task = PythonOperator(
                             task_id= 'pass_list',
                             provide_context=True,
                             python_callable=update_variable,
                             #op_kwargs={'input_file':file},
                             trigger_rule=TriggerRule.ALL_SUCCESS,
                             #provide_context = True,
                             #xcom_push=True,
                             dag=dag
                            )
    GCS_File_list >> python_task

如您所见，上面的脚本列出了存储桶中的文件，并将结果设置为气流变量.

as you see the above script lists files from a bucket and set the result to an airflow variable.

脚本2:导入脚本1中设置的变量值并在其上循环

script 2: import the variable value set in script 1 and loop over it

这是我可以获取变量值但无法对其进行迭代的地方

this is where i can get the variable value but cant iterate over it

脚本2:

#!/usr/bin/env python
from datetime import datetime, timedelta
from airflow import DAG
from airflowag.models import Variable
from airflow.operators.bash_operator import BashOperator
from airflow.utils.trigger_rule import TriggerRule
from airflow.operators import PythonOperator
from datetime import datetime

YESTERDAY = datetime.combine(
    datetime.today() - timedelta(days=1), datetime.min.time())
BQ_DATASET_NAME = 'abc'
CURRENT_TIME = datetime

files_list_str = Variable.get("files_list")
files_list = files_list_str.strip('][').split(',')
bucket = 'def'

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': YESTERDAY,
    'provide_context': True,
}

def gen_date(input_file,**kwargs):
# stepl1: check for file extension and remove it
idx_extension           = input_file.find(".")
input_file_name         = input_file[:idx_extension]
find_date_time_part     = re.findall("_(\d*?_\d*?_\d*)",input_file_name)
find_date_time_part     = str(find_date_time_part).split('_', 1)[-1].strip(']')
find_date_time_part     = str(find_date_time_part)
find_date_time_part     = re.sub("'",'', find_date_time_part)
find_date_time_part_len = len(find_date_time_part)

if find_date_time_part_len == 15:
    x = [a for a in find_date_time_part.split('_') if a]
    #get the date time part from the list 
    x = (' '.join(x[-2:]))
    #Using strptime to parse the string value as datetime object here our date format is YYYYMMDD hhmiss
    dt_obj = datetime.strptime(x, "%Y%m%d %H%M%S")
    # use strftime to format the date object into desired format in our case YYYY-MM-DD hh:mi:ss
    final_date_formatted = dt_obj.strftime("%Y-%m-%d %H:%M:%S")
    #print(type(find_date_time_part))
    return final_date_formatted
else:
    print("{}_{}".format(files_list,input_file))

with DAG('dag1', default_args=default_args,
schedule_interval=None
        ) as dag:
    for file in enumerate (files_list):
        Python_Task = PythonOperator(
                             task_id='pass_date',
                             provide_context=True,
                             python_callable=gen_date,
                             op_kwargs={'input_file':file},
                             trigger_rule=TriggerRule.ALL_SUCCESS,
                             provide_context = True,
                             #xcom_push=True,
                             dag=dag
                            )
        Python_Task

正如您在gen_date()函数中看到的那样，我正在else块中打印出变量名以及输入文件名

as you see in this function gen_date() i am printing out the variable name as well as the input file name in the else block

语句的输出 print("{} _ {}".format(files_list，input_file) 是['abc.csv'，'def.csv'] _ [

the output of the statement print("{}_{}".format(files_list,input_file) is ['abc.csv','def.csv']_[

我不确定为什么要传递"["而不是输入文件名 任何建议表示赞赏.

I am not sure why a "[" is getting passed instead of input_file name any advise appreciated.

我现在看到在for循环中，file_list被视为字符串而不是列表

如何使files_list成为列表而不是字符串.

how can i make the files_list as a list and not string.

推荐答案

因此，当我们获取气流变量(files_list)时，它会以字符串而不是列表的形式存储并返回，因此我们首先需要将变量转换为列表，然后然后循环访问该变量.

so when we get airflow variable (files_list) it is stored and returned as a string not a list so we would first need to convert the variable into a list and then loop on that variable.

这篇关于循环气流变量问题的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！