python - 即使 DAG 未运行,气流变量也会更新
问题描述
我正在从气流变量中读取一个整数变量,然后每次 DAG 运行时将该值加一并再次将其设置为该变量。
但是在下面的代码之后,每次刷新页面时,UI 上的变量都会发生变化。不知道是什么导致了这种行为
counter = Variable.get('counter')
s = BashOperator(
task_id='echo_start_variable',
bash_command='echo ' + counter,
dag=dag,
)
Variable.set("counter", int(counter) + 1)
sql_query = "SELECT * FROM UNNEST(SEQUENCE({start}, {end}))"
sql_query = sql_query.replace('{start}', start).replace('{end}', end)
submit_query = PythonOperator(
task_id='submit_athena_query',
python_callable=run_athena_query,
op_kwargs={'query': sql_query, 'db': 'db',
's3_output': 's3://s3-path/rohan/date=' + current_date + '/'},
dag=dag)
e = BashOperator(
task_id='echo_end_variable',
bash_command='echo ' + counter,
dag=dag,
)
s >> submit_query >> e
解决方案
每 30 秒 DAG 文件的气流进程(默认min_file_process_interval
设置)这意味着您拥有的任何顶级代码每 30 秒运行一次,因此Variable.set("counter", int(counter) + 1)
将导致变量计数器每 30 秒增加 1。
在顶级代码中与变量交互是一种不好的做法(不管增加值的问题)。它每 30 秒打开一个到 Metastore 数据库的连接,这可能会导致严重的问题并使数据库不堪重负。
要获取变量的值,您可以使用 Jinja:
e = BashOperator(
task_id='echo_end_variable',
bash_command='echo {{ var.value.counter }}',
dag=dag,
)
这是一种使用变量的安全方法,因为只有在执行运算符时才检索值。
如果您想将变量的值增加 1,请使用PythonOpeartor
:
def increase():
counter = Variable.get('counter')
Variable.set("counter", int(counter) + 1)
increase_op = PythonOperator(
task_id='increase_task',
python_callable=increase,
dag=dag)
只有当操作符运行时,python 可调用对象才会被执行。
推荐阅读
- python - 我们如何从链表中的第 n 个元素遍历到最后一个元素?
- kubernetes - 使用 prometheus pod 监控 golang webapp pod
- c++ - C++ openssl:编辑密码套件
- java - Android YoutubePlayerView 视频不可见
- python - 如何匹配具有不同时间分辨率的熊猫数据帧
- winapi - 使用 Windows CryptoUnprotectData 生锈 FFI
- c# - 如何测试一个非常简单的 AWS Lambda
- .htaccess - .htaccess 文件在嵌套目录中不起作用,但在父目录中起作用?
- android - 如何在 Android Studio 中使用 AsyncTask 从可绘制对象中设置图像
- ios - 如何唯一标识每个 UIView?