如何将变量传递给UI中的单个Airflow Task实例
Airflow是一个开源的任务调度和工作流管理平台，用于构建、调度和监控数据管道。在Airflow中，任务被组织成有向无环图（DAG），每个任务被称为一个Task。
要将变量传递给UI中的单个Airflow Task实例，可以通过以下步骤实现：
- 在Airflow中定义一个变量：在Airflow的Web界面中，点击"Admin" -> "Variables"，然后点击"Create"按钮创建一个新的变量。为变量指定一个名称和值，例如，变量名为"my_variable"，值为"my_value"。
- 在DAG文件中引用变量：在你的DAG文件中，导入
Variable类，并使用Variable.get()方法获取变量的值。例如，my_variable = Variable.get("my_variable")。 - 将变量传递给Task实例：在定义Task实例时，可以使用
params参数将变量传递给Task。例如，task = MyTask(task_id='my_task', params={'my_variable': my_variable})。 - 在Task中访问传递的变量：在Task的执行函数中，可以通过
self.params访问传递的变量。例如，my_variable = self.params.get('my_variable')。
通过以上步骤，你可以将变量传递给UI中的单个Airflow Task实例，并在Task中访问这些变量的值。
Airflow的优势在于其灵活性和可扩展性，它提供了丰富的功能和插件生态系统，可以满足各种复杂的数据管道需求。它适用于各种场景，包括数据处理、ETL流程、机器学习模型训练和部署等。
腾讯云提供了一系列与Airflow相关的产品和服务，包括云服务器、对象存储、数据库、容器服务等。你可以通过腾讯云的官方文档了解更多关于这些产品的详细信息和使用方法。
参考链接：
- Airflow官方文档：https://airflow.apache.org/
- 腾讯云产品文档：https://cloud.tencent.com/document/product/560
相关·内容
1回答
我有一个在成功和失败事件中执行动作和发送通知电子邮件的airflow作业，下面是我使用的代码。
#from builtins import range
from datetime import timedelta
from airflow.models import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.utils.dates import days_ago
from airflow.operators.email_operator import EmailOperator
from
浏览 2提问于2020-10-24得票数 1
我们希望在Dag中的Dag触发器期间从UI读取cli输入传递给dag。我尝试了下面的代码，但它不起作用。在这里，我将输入传递为{“kpi”：“ID123”}，并希望在函数get_data_from_bq中打印此ip值。
from airflow import DAG
from airflow.utils.dates import days_ago
from airflow.operators.python_operator import PythonOperator
from airflow import models
from airflow.models import Variabl
浏览 1提问于2021-12-30得票数 0
我正在尝试气流延迟操作。
import asyncio
from random import randint
from airflow import DAG
from airflow.sensors.base import BaseSensorOperator
from airflow.triggers.base import BaseTrigger, TriggerEvent
from work import default_args, default_start_date
class RandomTrigger(BaseTrigger):
def __init__(self):
浏览 9提问于2022-06-19得票数 0
我在windows 10 WSL上使用气流v2.0 (Ubuntu20.04)。警告信息是：
/home/jainri/.local/lib/python3.8/site-packages/airflow/models/dag.py:1342: PendingDeprecationWarning: The requested task could not be added to the DAG because a task with task_id create_tag_template_field_result is already in the DAG. Starting in Airfl
浏览 4提问于2020-10-13得票数 1
回答已采纳
我正在努力理解如何使用Airflow 2.0 DAG和任务装饰器读取任务中的dag配置参数。 考虑下面这个简单的DAG定义文件： from airflow.decorators import dag, task
from airflow.utils.dates import days_ago
@dag()
def lovely_dag():
@task(start_date=days_ago(1))
def task1():
return 1
something = task1()
my_dag = lovely_dag() 我可以使用UI
浏览 140提问于2021-01-13得票数 3
回答已采纳
我正在和BigQueryOperator一起试验气流。我想稍后我会使用，但我希望它首先在本地运行。我已经启动了气流，运行了一个BashOperator，我也可以运行airflow test <dag> <task>，其中task是我想要运行的大型查询任务，但是当我从UI触发DAG时，bigquery任务永远不会排队。相反，他们拥有REMOVED状态，什么都不会发生。
我的DAG定义如下：
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from dateti
浏览 0提问于2018-05-04得票数 11
2回答
我有以下两个文件。一个带有DAG和两个任务(DummyOperator和TaskGroup)。
# example_dag.py
from datetime import timedelta
from airflow import DAG
from airflow.operators.dummy import DummyOperator
from airflow.operators.python import PythonOperator
from airflow.utils.dates import days_ago
from includes.taskgroup import build_
浏览 8提问于2021-12-14得票数 0
1回答
我创建了一个变量(从Airflow UI)： Key: env_variables
Value: {'xx': 'yy`} 并尝试使用var.json.env_variables.xx内部bash操作符进行访问。但是"xx“的值是动态的，我是从REST API传递的，并使用dag_run.conf['param']访问它(dag_run.conf['param']应该返回xx)。最终，我想运行var.json.env_variables.xx。如何在Airflow中实现内部bash运算符？我正在尝试运行下面的代码，但它显示Jin
浏览 58提问于2021-09-23得票数 0
回答已采纳
我在Vault中使用气流1.10.10，并试图检索存储在Vault中的变量。我能够检索连接，但不能检索变量。
气流配置文件包含：
airflow.cfg
[secrets]
backend = airflow.contrib.secrets.hashicorp_vault.VaultBackend
backend_kwargs = {"connections_path": "connections", "variables_path": "variables", "mount_point": "secret
浏览 17提问于2022-03-25得票数 1
回答已采纳
在Airflow中，我试图在airflow中使用jinja模板，但问题是它没有被解析，而是被当作字符串处理。请参阅我的代码
from datetime import datetime
from airflow.operators.python_operator import PythonOperator
from airflow.models import DAG
def test_method(dag,network_id,schema_name):
print "Schema_name in test_method", schema_name
third
浏览 1提问于2018-11-26得票数 2
1回答
步骤(今日为2020年1月7日)
( 1)在气流中放置以下参数：
from datetime import datetime
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
dag = DAG(dag_id='example_dag', start_date=datetime(2020, 1, 1), catchup=False)
t1 = BashOperator(task_id='bash_task', bash_command='e
浏览 3提问于2020-01-07得票数 0
回答已采纳
2回答
我创建了一个定制的BashOperator，如下所示
from airflow.operators.bash_operator import BashOperator
class CustomOperator(BashOperator):
"""
Custom bash operator that just write whatever it is given as stmt
The actual operator is more complex
"""
def __init__(self, stm
浏览 2提问于2019-07-04得票数 2
1回答
这是我在dags文件夹中的数据文件。
Code that goes along with the Airflow located at:
http://airflow.readthedocs.org/en/latest/tutorial.html
"""
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator
from datetime im
浏览 0提问于2020-06-11得票数 0
回答已采纳
1回答
我在EC2上的一个码头集装箱里运行着气流。当我试图访问气流UI中的任务日志时，它将我重定向到localhost，然后我必须用EC2 ip替换localhost来查看日志。这是可行的，但如果我不必每次都更改地址，那就更好了。
我的问题是，我能改变这种行为吗?如果是的话，在哪里？
我在这个网站上搜索了一段时间，但我所能找到的只是日志本身有问题的人，但没有这个重定向问题。我以为我会在airflow.cfg上找到一些东西，但我不知道这是否是正确的地方。
这是我的airflow.cfg，如果需要，我可以发布任何其他文件：
[core]
# The home folder for airflow, def
浏览 0提问于2018-04-11得票数 10
回答已采纳
我无法从Python连接到Airflow的元数据数据库。
连接已设置，我可以使用UI的即席查询窗口查询元数据数据库。
如果我尝试使用相同的连接，但从Python，它将不会工作。
所以，详细地说，我有两个连接设置，都是从UI: connection 1: Conn type: MYSQL Host: airflow-sqlproxy-service Schema: composer-1-6-1-airflow-1-10-0-57315b5a登录: root
连接2:连接类型: MYSQL主机: 127.0.0.1架构: composer-1-6-1-airflow-1-10-0-57315b5a
浏览 0提问于2020-11-27得票数 1
当我运行我的管道并将日志写入S3时，我得到了以下错误。 Airflow版本: 1.10.14 我已经在Airflow Web UI中定义了s3_conn，以及密钥id和密码。 在airflow.cfg中，我有以下条目： base_log_folder = /root/airflow/logs
# Airflow can store logs remotely in AWS S3, Google Cloud Storage or Elastic Search.
# Set this to True if you want to enable remote lo
浏览 15提问于2020-12-22得票数 0
1回答
几个小时前，这段代码运行良好，但突然之间，我的伙伴们开始陷入“排队”状态。
下面是我试图运行的内容(我只是手动触发)：
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime as dt
def test_function():
print('Hello there')
default_args = {
'owner': 'airflow',
&#
浏览 3提问于2022-05-24得票数 0
@mock.patch.dict(
"os.environ",
AIRFLOW_CONN_LIVY_HOOK = "http://www.google.com",
clear= True
)
class TestLivyOperator(unittest.TestCase):
def setUp(self):
super().setUp()
self.dag = DAG(
dag_id = "test_livy",
defaul
浏览 5提问于2022-09-06得票数 0
回答已采纳
我正在尝试执行一个由几个Python函数组成的气流脚本。这些函数基本上是查询数据库并执行很少的任务。我正在尝试执行这是气流，这样我就能够单独监控每一个功能。给出的代码是我试图执行的代码，并得到以下错误
子任务: NameError:名称“task_instance”未定义为
## Third party Library Imports
import psycopg2
import airflow
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime
浏览 3提问于2018-05-30得票数 1
回答已采纳
从今天(2019-12-18)开始，我试图每5分钟运行一次DAG .我将开始日期定义为start_date:dt.datetime(2019, 12, 18, 10, 00, 00)，计划间隔定义为schedule_interval= '*/5 * * * *'。启动airflow scheduler时，没有看到任何任务正在运行。
但是，当我将start_date修改为start_date:dt.datetime(2019, 12, 17, 10, 00, 00) (即Yesterdays )时，DAG每10秒连续运行一次，而不是5分钟。
我认为解决这个问题的方法是正确地设置s
浏览 4提问于2019-12-18得票数 1
回答已采纳
我在脚本下面有一个气流，它作为一个函数运行所有python脚本。我希望每个python函数都能单独运行，这样我就可以跟踪每个函数及其状态。
## Third party Library Imports
import psycopg2
import airflow
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
#from airflow.operators.bash_operator import BashOperator
from datetime import dat
浏览 3提问于2018-05-28得票数 0
2回答
到目前为止，我正试着熟悉气流，并热爱它。
但是，有一件事我有点不清楚，那就是如何正确地为我的守护进程提供参数，我想在其中运行相同的进程，但是并行地运行多个业务(lob)。因此，基本上，我希望在每次运行中运行多个lob的下面的dag，并让每个lob运行在parralel中。
因此，假设我定义了一个变量，它是一个lobs数组，如"lob1“、"lob2”等等。我想将下面bigquery语句中的'mylob‘替换为'lob1’，然后是'lob2‘等等。
我在想，也许我可以将lobs作为一个变量从ui中存储起来，然后在后台循环，但是我不确定这是否会是连续的，因为
浏览 3提问于2017-03-21得票数 7
回答已采纳
我有一个airflow dag，而我要做的是读取存储在airflow UI中的变量(用户名和密码)，并将这些变量值作为导出的值传递到操作系统中。原因是因为我使用的是dbt yml文件，它要求我读取环境变量'dbt_user‘。(另一种方法是在yaml文件中设置密码，这是不安全的。 default:
target: dev
outputs:
dev:
type: snowflake
account: xxxx
user: "{{ env_var('dbt_user') }}" 我尝试编写一个dag来执
浏览 114提问于2021-11-04得票数 0
我是Apache气流的新手，我试图编写我的第一个Dag，它有一个基于另一个任务的任务(使用ti.xcom_pull)
PS :我使用VScode在WSL Ubuntu20.04中运行气流。
我创建了一个任务1 (task_id = "get_datetime")，该任务运行"date“bash命令(它可以工作)
然后我创建了另一个任务(task_id='process_datetime')，它占用第一个任务的日期时间并对其进行处理，我设置了python_callable，一切都很好。
问题是，当我在终端中运行“气流任务测试(ti.xcom_pull ta
浏览 12提问于2022-11-04得票数 0
2回答
是否可以通过UI将参数传递给Airflow的作业？
AFAIK，DAG中的'params‘参数是在python代码中定义的，因此不能在运行时更改。
浏览 6提问于2017-11-21得票数 18
回答已采纳
我正在与virtualenv合作。我正在尝试使用DAG文件夹中的包。airflow_home目录的当前状态为：
airflow_home/airflow.cfg
airflow_home/airflow.db
airflow_home/dags/__init__.py
airflow_home/dags/hello_world.py
airflow_home/dags/support/inner.py
airflow_home/dags/support/__init__.py
hello_world.py有代码：
from datetime import datetime
from air
浏览 1提问于2017-08-20得票数 15
有没有办法将命令行参数传递给Airflow BashOperator。目前，我有一个python脚本，它接受日期参数并执行一些特定的活动，比如清理比给定日期旧的特定文件夹。
在只有一个任务的简化代码中，我想要做的是
from __future__ import print_function
from airflow.operators import BashOperator
from airflow.models import DAG
from datetime import datetime, timedelta
default_args = {
'owner'
浏览 5提问于2017-02-03得票数 10
2回答
我正在尝试使用python代码中的模板并按照定义的使用globals()来生成airflow dags
定义dag对象并保存。下面是我的代码：
import datetime as dt
import sys
import airflow
from airflow.models import DAG
from airflow.operators.bash_operator import BashOperator
argumentList = sys.argv
owner = argumentList[1]
dag_name = argumentList[2]
taskID = argum
浏览 9提问于2020-06-04得票数 0
2回答
我在风流v2.3.0上使用了taskflow API。
我有一个DAG它有很多任务。第一个任务(TaskA)打开一个EC2实例，最后一个任务(TaskX)关闭EC2实例。
TaskA返回EC2实例的实例ID，然后TaskC使用它在两个任务之间关闭它，有任务B、C、D、...W，任何任务都可能失败。当这些任务中的任何一个失败时，我希望触发TaskX来关闭我的EC2实例。
我知道我可以在on_failure_callback中使用@dag，但问题是TaskX需要一个在进行DAG定义时不知道的EC2实例ID --它需要从TaskA的返回值中获取(参见下面的MCVE代码)。
@dag(on_failu
浏览 11提问于2022-09-16得票数 0
回答已采纳
1回答
我使用的是与Kubernetes执行器一起的气流，并在本地进行测试(使用minikube)，虽然我能够启动并运行它，但我似乎无法将我的日志存储在S3中。我已经尝试了所有被描述的解决方案，我仍然得到了以下错误，
*** Log file does not exist: /usr/local/airflow/logs/example_python_operator/print_the_context/2020-03-30T16:02:41.521194+00:00/1.log
*** Fetching from: http://examplepythonoperatorprintthecontex
浏览 3提问于2020-03-30得票数 3
回答已采纳
我目前使用的是Airflow (版本: 1.10.10)， 我对创建DAG很感兴趣，它将每小时运行一次， 这将带来Docker容器的使用信息(磁盘使用)， (可通过docker CLI命令( df -h)获得的信息)。 我的理解是：“如果xcom_push为真，那么当bash命令完成时，写入标准输出的最后一行也将被推送到XCom。” 但我的目标是从bash命令中获得一个特定值，而不是最后编写的一行。 例如，我想得到这一行(参见screeeshot) "tmpfs 6.2G 0 6.2G 0% /sys/fs/cgroup“ 转换为我的XCOM值，这样我就可以编辑和提取其中的特定值，
浏览 40提问于2021-10-31得票数 0
回答已采纳
我使用CLI创建了一个自定义包(使用Click构建)。这个包可以做两件事:运行预处理和运行机器学习模型。我创建了此客户包的Docker镜像，并将其推送到AWS (ECR)上的私有注册表。 现在我想用Airflow运行这个容器，我想在一个EC2实例上运行它。我正在用docker-compose运行它。 在本例中，我将只关注一项任务:运行容器进行预处理。 然而，现在我得到了‘上游失败’的t2。 from datetime import timedelta
import airflow
from airflow import DAG
from airflow.operators.bash_oper
浏览 57提问于2021-09-26得票数 2
我知道如何使用变量动态地运行DAG的任务，在为同一个DAG触发多次运行之前，它运行得很好。
也就是说，一旦在data/to/load/ dir下创建了一个包含文件的新目录，我在某个地方编写了一个脚本，它将触发airflow variables -set dir data/to/load/$newDir，然后是airflow trigger_dag dyn_test。现在，假设在data/to/load/下创建目录"a“和"b”(类似的时间)，这将使airflow variable + airflow trigger_dag调用两次，变量集调用上有两个不同的输入(一个后缀为
浏览 4提问于2019-04-14得票数 1
我正在运行空气流量2.1.4使用码头-合成和芹菜执行器。到目前为止，我已经能够从芹菜工作容器启动和运行简单的DockerOperator任务，但是现在当我尝试将一个目录从共享驱动器挂载到任务容器时，我得到了一个错误(下面的日志文件)。如果我不定义挂载参数，Dag工作得很好。因此，我猜有些信息或特权不会传递给由芹菜工人容器创建的容器。有什么建议吗?接下来要尝试什么？
DAG档案：
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetim
浏览 10提问于2021-11-03得票数 0
回答已采纳
要求DAG一个接一个地运行，并且每个DAG都成功运行。
我有一个主DAG，在这个主DAG中，我调用所有DAG来依次执行
在每个dag_A、dag_B、dag_C中，我都必须给schedule_interval = None，并在GUI中手动打开
我正在使用ExternalTaskSensor，因为即使在第一个dag_A中的所有任务都完成之前，它就启动了第二个dag_B，为了避免这样的问题，我正在使用ExternalTaskSensor.If更好的实现，请告诉我。
不知道我在这里错过了什么
代码: master_dag.py
import datetime
import os
from date
浏览 5提问于2020-06-25得票数 3
5回答
绕流变量问题
、、、
我很难在脚本中的一个气流变量上循环。
因此，我需要列出桶中以字符串为前缀的所有文件。
下一个循环并执行一些操作。
我试着使用xcomm和subdag，但是我搞不懂，所以我想出了一种新的方法。
不过，它涉及两个脚本
使用我生成的值设置气流变量。
下面是密码。
#!/usr/bin/env python
with DAG('Test_variable', default_args=default_args,
schedule_interval=None
) as dag:
GCS_File_list = GoogleCloudStorageListOperato
浏览 0提问于2019-09-12得票数 0
我刚在一个服务器上安装了气流1.10
sudo -E pip-3.6 install apache-airflow[celery,devel,postgres]
在那之后我可能也做了这个
sudo -E pip-3.6 install apache-airflow[all]
但是无论如何，当我运行airflow version时，我会得到以下输出
[ec2-user@ip-1-2-3-4 ~]$ airflow version
[2018-08-29 16:09:59,088] {{__init__.py:51}} INFO - Using executor SequentialExecuto
浏览 0提问于2018-08-29得票数 8
1回答
我是阿帕奇气流公司的新手。互联网上有很多基本的DAG例子。不幸的是，我没有找到任何单任务DAG的例子。
DAG的大多数示例都在.py脚本的末尾包含位移位操作符，后者定义了任务顺序。例如：
# ...our DAG's code...
task1 >> task2 >> task3
但是如果我的DAG目前只有一个任务呢？我的问题是-我是否需要在Python文件的末尾使用这个任务的名称？或者，如果我们在范围内只有一个任务，气流将处理它本身，下面的最后一行代码是多余的？
from datetime import timedelta
from airflow.opera
浏览 2提问于2021-09-08得票数 7
回答已采纳
我是airflow的新手，当我单击Task Instance上下文菜单上的run 'ignore all dependence‘时，如下所示：
它会导致“仅适用于CeleryExecutor”
我尝试刷新Web UI，但它不起作用。
(我使用LocalExecutor，不想使用CeleryExecutor)
为什么会发生这种情况?当我使用LocalExecutor时，如何运行单个任务忽略对Web的所有依赖
浏览 0提问于2017-06-09得票数 8
1回答
我正在尝试遵循Airflow 在DAG中使用变量。
这些变量是在Airflow UI中设置的，我使用这些变量创建了一个简单的DAG。加载文档中所示的变量会导致以下错误：
[2021-02-18 16:32:15,350] {variable.py:64} ERROR - Can't decrypt _val for key=foo, invalid token or value
[2021-02-18 16:32:15,364] {taskinstance.py:1396} ERROR - 'Variable foo does not exist'
下面是我正在使用的设
浏览 7提问于2021-02-19得票数 2
回答已采纳
我正在尝试从CLI触发气流DAG，DAG触发正常，但我收到一堆python警告消息，谁能帮助我，如何抑制这些警告消息。 命令：airflow trigger_dag -e 2021-07-13 Test_CI_DAG 日志： /var/xxx/airflow/venv/lib/python2.7/site-packages/airflow/contrib/operators/ssh_operator.py:65: PendingDeprecationWarning:
Invalid arguments were passed to SSHOperator (task_id: run_spar
浏览 31提问于2021-07-13得票数 0
1回答
写入气流2台
、、
我已经在气流1.10.14很长一段时间了，现在我正在尝试升级到气流2.4.3 (最新？)我已经在新的格式中构建了这个数据库，希望能吸收语言，了解新格式的工作原理。下面是我的爸爸：
from airflow.decorators import dag, task
from airflow.models import Variable
from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator
from airflow.providers.microsoft.mssql.opera
浏览 9提问于2022-11-17得票数 1
回答已采纳
我对气流很新，但我真的很难理解如何传递小的xcom值。有时，这似乎没有问题，但有时，它需要我几个小时。这是一个让我非常兴奋的例子。
我试图触发一个外部DAG，我需要将一个数字作为param传递给外部dag。我不知道这个号码，但我是从我爸爸的任务中得到的。如何将其作为conf/param传递给触发的DAG？
这可能最容易通过一个例子来理解：
import datetime
from airflow import DAG
from airflow.decorators import task
from airflow.models import Param
from airflow.operat
浏览 7提问于2022-09-10得票数 0
我在Airflow 1.10上有一个奇怪的bug。我想试着发送电子邮件和微软团队的通知。我做了一个小的哑巴DAG来试一下。一切运行正常，但我连续收到两个通知。团队中的2封电子邮件和2条消息。 我在团队中使用了这个：https://github.com/mendhak/Airflow-MS-Teams-Operator 这里是Dags： from datetime import datetime
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.ope
浏览 34提问于2020-06-17得票数 4
2回答
您好，我刚刚开始使用Airflow，但我无法在快速入门run：airflow run example_bash_operator runme_0 2015-01-01中完成任务。
我刚刚使用Python2.7.6创建了一个conda环境，并通过安装了airflow==1.8.0的pip安装了airflow。然后，我运行了中列出的命令。
当我尝试运行第一个任务实例时，通过查看UI，似乎什么都没有发生。以下是该命令的输出：
(airflow) ✔ se7entyse7en in ~/Projects/airflow $ airflow run example_bash_operator ru
浏览 0提问于2017-07-28得票数 3
为了进一步执行，我需要在另一个任务中使用来自oracleOperator的输出。我遇到的麻烦是，当我把数据拉到另一个任务中并打印出来时，它会给出一个零的结果。没有抛出错误，但没有传递数据。此外，任务UI中的xcom选项卡显示键和值的空白。
我的代码如下：
from airflow import DAG
from airflow.operators.oracle_operator import OracleOperator
from airflow.operators.python_operator import PythonOperator
from airflow.utils.dates i
浏览 3提问于2020-03-23得票数 3
我有一个尝试连接到http端点的dag，但不知怎么的，它不起作用。我使用envinorment变量定义了连接，但是http传感器没有看到它，而且我的错误程度低于错误，但是该变量是在系统中创建的。这里出什么问题了？下面是dag和完整的错误代码。
The conn_id `AIRFLOW_VAR_FOO` isn't defined
DAG:
import os
import json
import pprint
import datetime
import requests
from airflow.models import DAG
from airflow.operators.py
浏览 17提问于2022-11-04得票数 0
回答已采纳
3回答
关于“动态任务”的其他问题似乎解决了在计划或设计时动态构建DAG的问题。我对在执行过程中将任务动态添加到DAG很感兴趣。
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator
from datetime import datetime
dag = DAG('test_dag', description='a test',
浏览 2提问于2018-02-03得票数 21
回答已采纳
我试图在气流中运行一个DAG，将数据集摄取到google云存储中。这是DAG脚本：
import os
from airflow import DAG
from airflow.utils.dates import days_ago
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator
from google.cloud import storage
from airflow.providers.google.cloud.operators.bi
浏览 26提问于2022-09-28得票数 1
1回答
我安装了气流，使用对接-合成和运行db init命令。我正在尝试创建一个使用DockerOperator执行某些脚本的DAG。在我的DockerOperator中，脚本试图读取气流变量并使用BaseHook获取连接。但是，我的DockerOperator中的脚本似乎连接到(空的) sqlalchemy数据库，而不是通过UI设置填充连接和变量的初始化postgres数据库。
是否有一种方法可以将DockerOperator上下文提供给包含通过UI设置的连接和变量的气流数据库？
我的气流DAG
from datetime import datetime
from json import dumps
浏览 6提问于2022-06-02得票数 1
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
