如何获取airflow dag中运行的任务列表
Airflow是一个开源的任务调度和工作流管理平台，可以用于构建、调度和监控复杂的数据管道和工作流。在Airflow中，DAG（Directed Acyclic Graph）是任务的有向无环图，用于定义任务之间的依赖关系和执行顺序。
要获取Airflow DAG中运行的任务列表，可以按照以下步骤进行操作：
- 登录到Airflow的Web界面，通常通过访问Airflow的URL来实现。具体的URL地址可以根据部署环境进行调整。
- 在Airflow的Web界面中，导航到"DAGs"或"任务调度"等相关选项卡，这取决于Airflow的版本和配置。
- 在"DAGs"或"任务调度"页面中，您将看到已定义的所有DAG。选择您感兴趣的DAG，点击进入该DAG的详细信息页面。
- 在DAG的详细信息页面中，您将看到该DAG中定义的所有任务。这些任务通常以任务名称或任务ID的形式列出。
- 浏览任务列表，您可以获取每个任务的名称、描述、依赖关系以及其他相关信息。根据需要，您还可以查看任务的执行状态、日志和历史记录等。
请注意，以上步骤仅适用于Airflow的基本用法。具体的操作和界面可能因Airflow的版本和配置而有所不同。此外，您还可以通过Airflow的命令行界面（CLI）或API来获取任务列表的信息。
推荐的腾讯云相关产品：腾讯云容器服务（Tencent Kubernetes Engine，TKE），腾讯云函数计算（Tencent Cloud Function），腾讯云数据工厂（Tencent Data Factory）。
腾讯云容器服务（TKE）：https://cloud.tencent.com/product/tke 腾讯云函数计算（Tencent Cloud Function）：https://cloud.tencent.com/product/scf 腾讯云数据工厂（Tencent Data Factory）：https://cloud.tencent.com/product/df
相关·内容
如何获取在具有dag-id的airflow dag中运行的任务列表？任务之间的连接并不重要。谢谢
假设我有来自"airflow list_dags“的dag-id
浏览 94提问于2019-07-30得票数 2
我知道以前有人问过这个问题，但我想把它说得很简单。我正在使用Astronomer运行Airflow v1.10.10。每个dag运行(每天)可以有不同数量的任务。此任务数将由给定dag运行(即当天)的数据库表中的行数定义。
我将在dag文件中运行一个查询，并创建一个如下列表：
list = client.query(SELECT * FROM TABLE WHERE DATE = {{ds}})
然后，我将根据列表的大小创建任务：
for i in range(1, len(list)):
sample_airflow_operator(name=i + '_operator&
浏览 2提问于2021-06-03得票数 0
我需要重新启动气流。我想确保我在它空闲的时候做，这样我就不会因为重新启动气流的worker组件而中断作业。 如何查看正在运行的DAG？ 我在UI中看不到任何列出当前正在运行的DAG的内容。 我在airflow命令行界面中看不到任何用于列出当前正在运行的DAG的命令。 我找到了可以连接到数据库的airflow shell，但我对Airflow内部结构的了解还不够，不知道在哪里可以看到正在运行的内容。
浏览 37提问于2020-10-20得票数 0
回答已采纳
3回答
我想使用一个脚本生成多个气流dags。dag名称应为"test_parameter“。下面是我的脚本：
from datetime import datetime
# Importing Airflow modules
from airflow.models import DAG
from airflow.operators import DummyOperator
# Specifying the default arguments for the DAG
default_args = {
'owner': 'Test',
'
浏览 2提问于2018-03-17得票数 6
我有一个正在运行的dag，我可以使用REST API停止它吗？ 有一种方法可以删除正在运行的dag：https://airflow.apache.org/docs/apache-airflow/stable/stable-rest-api-ref.html#operation/delete_dag_run。 但是没有REST API如何停止一个正在运行的dag？ 谢谢!
浏览 145提问于2021-10-09得票数 1
回答已采纳
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使用芹菜执行程序，但我想使用Kubernetes Executor运行一个特定的dag。我无法推断出一个好的和可靠的方法来实现这一点。
我有一个声明要使用CeleryExecutor的airflow.cfg。我不想改变它，因为除了一个dags之外，所有的dags都需要它。
# The executor class that airflow should use. Choices include
# SequentialExecutor, LocalExecutor, CeleryExecutor
executor = CeleryExecutor
我的dag代码：
from da
浏览 2提问于2019-05-27得票数 6
我有一个达格，每小时安排一次。假设凌晨1:00，凌晨2:00，凌晨3:00假设02:00被选中，但是如果01:00am dag运行仍在进行中，则需要取消02:00am实例。
我在尝试这个密码。
local_tz = pendulum.timezone("America/Chicago")
default_args = {
'owner': 'airflow',
'depends_on_past': True,
'start_date': datetime(2020, 11, 15, tzi
浏览 15提问于2020-11-17得票数 0
回答已采纳
我使用root帐户在集群上安装了Airflow。我知道这是不好的练习，但它只是测试环境。我创建了一个简单的DAG：
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedelta
dag = DAG('create_directory', description='simple create directory workflow', start_date=datetime(2017,
浏览 9提问于2017-10-30得票数 3
回答已采纳
我们有3个composer GKE节点和3个worker pods均匀分布在所有3个节点上。我需要知道如何检查哪个DAG或任务当前在哪个pod中运行。我试着运行airflow list_dag，但我想它显示了所有的dags。我只需要知道哪个dag在哪个pod中运行。是否可以将pod从一个节点移动到另一个节点。有时我的pod并不均匀地分布在所有3个节点上。
浏览 0提问于2021-01-12得票数 0
在实际的DAG中，我需要首先获得一个ID列表，然后对每个ID运行一组任务。
我已经使用动态任务映射将列表传递给单个任务或操作符，让它处理列表，但是我们也可以使用TaskGroup来完成这个任务吗？
如果我能够知道如何在TaskGroup级别传递变量值，那么它将在所有子任务中使用该值，那么我应该能够满足我的要求。
以下应该给你一个想法，我正在寻找什么，只是需要帮助，使它发挥作用。
from airflow import DAG, XComArg
from datetime import datetime
from airflow.decorators import task
from airfl
浏览 5提问于2022-10-29得票数 1
如果我运行命令airflow list_dags，它将返回一个DAG列表。但是，DAG列表与我在web界面中DAG选项卡下看到的DAG列表不同。airflow list_dags命令仅返回教程DAGs，而不返回airflow/dags目录中的DAGs。
为什么airflow为我提供了两个不同的DAG列表？
我运行的是Airflow 1.8.1。
浏览 0提问于2017-08-01得票数 2
我使用Apache气流，我希望它发送邮件通知sla错过。我将电子邮件嵌入存储为气流变量，我有一个进程，它的任务之一是使用EmailOperator发送电子邮件。
现在出现了这个问题，因为无论如何，当我的发送邮件任务运行到所有收件人时，它都只向列表中的第一个入口发送sla错过通知，在我的示例中，这意味着test1@test.com。
这是个bug，还是为什么它不起作用？
下面是我的dag和airlfow变量：
from airflow import DAG
from datetime import datetime, timedelta
from airflow.operators.ema
浏览 2提问于2018-09-05得票数 1
回答已采纳
我已经创建了一个dag，该dag在Airflow UI中可用，并将其打开以运行它。运行dag后，状态显示为可重试。在那之后，我去了服务器，使用了命令"Airflow scheduler“，然后dag就成功了。
在运行dag之前，调度程序已经启动并运行，我不确定为什么会发生这种情况。当我们创建一个新的dag时，我们需要运行气流调度器吗？想知道调度器是如何工作的。
谢谢
浏览 0提问于2020-05-20得票数 1
当我在终端中运行它时，下面的代码运行正常：
data = []
def _step_one():
for i in range(10):
data.append(i)
def _step_two():
print(f'data: {data}')
_step_one()
_step_two()
它打印data: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
但是，当我在Apache Airflow上运行同样的命令时，第一次初始化时data是空的[]
from airflow import DAG
from airflow.o
浏览 9提问于2021-02-24得票数 0
回答已采纳
我有一个DAG，它有一个步骤read_date_information_file，它读取一个文件并返回查询列表(我可以从输出中访问)。然后，我希望遍历此列表，并使用AWSAthenaOperator对此列表中的每个查询执行雅典娜上的查询。 def get_date_information(ti):
s3 = boto3.client('s3')
data = s3.get_object(Bucket=output_bucket, Key=key)
contents = data['Body'].read().decode("ut
浏览 86提问于2021-09-22得票数 0
我正试图从另一个人身上触发另一个人。我也在使用TriggerDagRunOperator。
我有以下两条线索。
Dag 1:
from datetime import datetime
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator
from airflow.operators.dagrun_operator import TriggerDagRunOper
浏览 4提问于2020-01-06得票数 0
我知道如何使用变量动态地运行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
我正在和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
是否有办法对非硬编码列表中的每个项目执行气流操作？对于动态计算列表中的每个项(@task的结果)，希望以环境变量的形式传递该项，以运行一个对接者映像：
for item in list_fetched_from_a_task:
run docker container passing in the item as a environment variable
此DAG将在sql查询的列表结果中打印每个项：
from airflow import DAG
from airflow.decorators import task
from datetime import datetime
f
浏览 6提问于2022-10-29得票数 0
回答已采纳
1回答
def skip_update_job_pod_name(dag):
"""
:param dag: Airflow DAG
:return: Dummy operator to skip update pod name
"""
return DummyOperator(task_id="skip_update_job_pod_name", dag=dag)
def update_pod_name_branch_operator(dag: DAG, job_id: str):
&
浏览 1提问于2022-07-08得票数 1
回答已采纳
在ECS Fargate上运行Airflow实例。问题是我无法在DAG中运行代码来调用现有的Glue Job。下面是DAG脚本。 import boto3
import os
import logging
import time
import sys
import botocore
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from airflow.operators.bash_ope
浏览 21提问于2021-01-08得票数 0
回答已采纳
3回答
我已经安排了我的气流DAG运行，每个DAG都有一个任务在其中。当DAG运行时，它们中的任务不会被执行。
下面是我为同一个编写的代码(我正在尝试将SSH放入EC2服务器并运行bash命令)
from datetime import timedelta, datetime
from airflow import DAG
from airflow.contrib.operators.ssh_operator import SSHOperator
default_args = {
'owner': 'airflow',
'depends
浏览 0提问于2019-07-05得票数 2
在一个进程运行期间读取一个dag定义文件的频率是多少？
拥有一个需要很长时间构建的大型守护进程(~1-3分钟)。当进程运行时，查看每个任务的日志，似乎每个任务在运行之前都在执行dag定义文件.
*** Reading local file: /home/airflow/airflow/logs/mydag/mytask/2020-01-30T04:51:34.621883+00:00/1.log
[2020-01-29 19:02:10,844] {taskinstance.py:655} INFO - Dependencies all met for <TaskInstance: my
浏览 4提问于2020-01-30得票数 1
回答已采纳
我目前正在试验一种新的概念，即操作员将与外部服务通信以运行操作员，而不是在本地运行操作员，外部服务可以与气流通信以更新DAG的进度。
例如，假设我们有一个bash操作符：
bash_task = BashOperator(
task_id="bash_task",
bash_command="echo \"This Message Shouldn't Run Locally on Airflow\"",
)
这是DAG的一部分：
from airflow import DAG
from airflow.operators.
浏览 2提问于2022-08-10得票数 0
回答已采纳
1回答
下面是我创建的DAG的简单复制。DAG具有分支运算符，用于选择合并为公共任务的执行流程。该任务应该生成一个文件列表，该列表将用于为列表文件中的每个条目创建一个任务。问题是我不能让动态任务执行。 """
Required packages to execute DAG
"""
from __future__ import print_function
from builtins import range
import airflow
from airflow.models import DAG
from datetime import date
浏览 14提问于2019-01-09得票数 2
回答已采纳
我在运行Apache气流1.8.1。我希望在我的实例上运行32多个并发任务，但无法使任何配置工作。
我使用的是CeleryExecutor，UI中的气流配置为parallelism和dag_concurrency显示了64，我已经多次重新启动了气流调度程序、web服务器和工作人员(我实际上正在Vagrant中本地测试这个配置，但也在一个EC2实例上进行了测试)。
airflow.cfg
# The amount of parallelism as a setting to the executor. This defines
# the max number of task instances
浏览 0提问于2018-12-05得票数 4
回答已采纳
在触发dag运行时，我尝试将以下配置参数传递给Airflow CLI。以下是我正在使用的trigger_dag命令。
airflow trigger_dag -c '{"account_list":"[1,2,3,4,5]", "start_date":"2016-04-25"}' insights_assembly_9900
我的问题是，如何在dag运行中访问传递到操作符内部的con参数。
浏览 11提问于2017-04-27得票数 24
回答已采纳
3回答
在我的一些Apache Airflow安装中，计划运行的DAG或任务即使在调度程序似乎没有完全加载时也不会运行。如何增加可以并发运行的DAG或任务的数量？
类似地，如果我的安装处于高负载下，并且我想限制我的Airflow工作人员拉取排队任务的速度(例如减少资源消耗)，我可以调整什么来降低平均负载？
浏览 2提问于2019-05-30得票数 45
回答已采纳
1回答
我正在Ubuntu20.04中运行气流2.1.0，我在一个进程中检测到了一种优化的可能性，我需要帮助找出如何改进它。提前谢谢你。
我正在运行一个进程来执行两种任务：
Raw层=查询S3 bucketTrusted层=从S3原始桶中读取的数据库和记录文件，以及S3可信层桶中的记录
我需要允许每个任务组的可信任务在Raw任务结束后立即执行，但当发生这种情况时，我还需要触发下一个任务组的Raw任务。由于受信任的层不使用数据库，所以我可以通过允许它已经启动下一个原始任务来优化dag执行时间。也不能同时运行查询，所以只能一次运行一个原始任务。
实际流程：(未优化，原始任务本来可以启动时处于空闲状态)
T
浏览 7提问于2022-04-13得票数 0
1回答
我试图在测试环境中使用多个任务来测试一个守护进程。我能够测试与dag关联的单个任务，但我希望在dag中创建多个任务，并启动第一个任务。用于测试我正在使用的dag中的一个任务。
task1.run()
正在被处决。但是，当我在后台的下游有一个接一个的任务时，同样的情况就不起作用了。
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedelta
default_args = {
'owner'
浏览 0提问于2018-04-24得票数 13
我有一个气流DAG，我需要从气流上下文中获得DAG触发的参数。
以前，我有在DAG步骤中获取这些参数的代码(我使用的是气流2中的Taskflow API ) --类似于这样：
from typing import Dict, Any, List
from airflow.decorators import dag, task
from airflow.operators.python import get_current_context
from airflow.utils.dates import days_ago
default_args = {"owner": "
浏览 5提问于2021-10-21得票数 3
回答已采纳
1回答
我正在尝试创建动态DAG，然后将它们发送到调度程序。我尝试了的引用，它运行得很好。我对它做了一些修改，如下代码所示。在调试问题时需要帮助。
我试过了。试着运行文件。Dag被执行，globals()正在打印所有的DAGs对象。但不知何故没有在list_dags或UI中列出
from datetime import datetime, timedelta
import requests
import json
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from airfl
浏览 1提问于2019-08-12得票数 0
回答已采纳
2回答
我对气流完全陌生。我想在指定的日期运行一个简单的DAG。我正在努力改变开始日期、执行日期和回填之间的差异。运行DAG的命令是什么？
以下是我尝试过的：
airflow run dag_1 task_1 2017-1-23
当我第一次运行该命令时，任务正确地执行，但是当我再次尝试时，它没有工作。
下面是我运行的另一个命令：
airflow backfill dag_1 -s 2017-1-23 -e 2017-1-24
我不知道这个命令能带来什么结果。day会从23天执行到24天吗？
在运行上述两个命令之前，我执行了以下操作：
airflow initdb
airflow scheduler
浏览 7提问于2017-01-23得票数 20
回答已采纳
考虑下面的DAG示例，其中第一个任务get_id_creds从数据库中提取凭据列表。这个操作告诉我数据库中的哪些用户能够运行进一步的数据预处理，并将这些ids写入文件/tmp/ids.txt。然后，我将这些in扫描到我的DAG中，并使用它们生成可以并行运行的upload_transaction任务列表。
我的问题是:是否有一种用气流来做这件事的更准确、更动态的方法？我在这里的感觉既笨拙又脆弱。如何将有效ID的列表从一个进程直接传递给定义后续下游进程的进程？
from datetime import datetime, timedelta
import os
import sys
from a
浏览 8提问于2017-07-25得票数 7
我一直在尝试在气流中使用“example_subdag_operator”。在我将“start_date”更改为“datetime.now()”之后，然后手动触发一个dag运行。操作符是绿色的，但是当我放大到子程序本身时，没有执行任何操作(在图形视图中是白色的)，“Run”下拉列表是空的。\
我用师父制作的最新版本。我还测试了pip发布的版本。仍然是同样的问题\
这是个虫子吗？还是我漏掉了什么？
from datetime import datetime
from airflow.models import DAG
from airflow.operators.dummy_operator
浏览 11提问于2017-02-06得票数 0
我正在写一个气流DAG，有一些函数的问题。我尝试通过将数据打印到标准输出并使用logging库来进行调试。
我的DAG示例如下：
from datetime import timedelta
import airflow
import logging
from airflow.models import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.contrib.hooks.datadog_hook import Dat
浏览 45提问于2019-08-08得票数 7
我在airflow中创建了一个DAG，该DAG使用测试和回填命令成功运行，但当我从airflow UI手动运行它时，它会出错。有什么办法解决我的问题吗？注意，DAG在LocalExecutor中运行。
在UI Airflow中记录错误：
*** Log file isn't local.
*** Fetching here: http://:8793/log/test/start/2017-10-31T09:14:32.378718
*** Failed to fetch log file from worker.
*** Reading remote logs...
*** Uns
浏览 0提问于2017-10-31得票数 1
我制作了一个非常简单的DAG，如下所示：
from datetime import datetime
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
cleanup_command = "/home/ubuntu/airflow/dags/scripts/log_cleanup/log_cleanup.sh "
dag = DAG(
'log_cleanup',
description='DAG for deleting old logs
浏览 1提问于2018-04-02得票数 2
回答已采纳
我正在尝试基于dagrun输入变量设置S3Key传感器的bucket_key。我有一个dag "dag_trigger“，它使用TriggerDagRunOperator来为dag "dag_triggered”触发dagrun。我试图扩展示例。
因此，我希望发送一个变量来触发dag，并根据该变量的值在S3KeySensor任务中设置S3KeySensor值。我知道如何在PythonOperator可调用函数中使用variable变量，但不知道如何在传感器对象上使用它。
dag_trigger dag：
import datetime
from airflow import
浏览 0提问于2018-02-08得票数 5
回答已采纳
我需要创建一个airflow操作符，它接受几个输入，并返回一个字符串，该字符串将用作下一次运行的另一个操作符的输入。我对airflow dags和操作员是新手，对如何正确地这样做感到困惑。由于我是为使用airflow和构建dag的人构建此应用程序，而我不是一个实际的airflow用户或dag开发人员，因此我想获得有关正确使用它的建议。我已经创建了一个运算符，它返回一个令牌(只是一个字符串，所以hello world运算符示例可以很好地工作)。这样做，我看到了用于dag执行的XCOM值。但是如何正确地检索该值并将其输入到下一个运算符？在我的示例中，我只是调用了相同的运算符，但实际上它将调用一个
浏览 26提问于2021-08-14得票数 0
回答已采纳
5回答
我正在尝试使用airflow来定义我想要从命令行手动触发的特定工作流。
我创建了DAG并添加了一系列任务。
dag = airflow.DAG(
"DAG_NAME",
start_date=datetime(2015, 1, 1),
schedule_interval=None,
default_args=args)
然后我在终端中运行
airflow trigger_dag DAG_NAME
什么都不会发生。调度程序正在另一个线程中运行。任何方向都是非常受欢迎的。谢谢
浏览 2提问于2016-05-05得票数 24
2回答
我有以下气流代码：
dag = DAG(
dag_id='example_python_operator', default_args=args,
schedule_interval=None)
for i in range(5):
t = BashOperator(
task_id = 'some_id',
params={'i':i},
bash_command = "python airflow/scripts/a.py '{{ execution_da
浏览 0提问于2018-06-12得票数 0
回答已采纳
我想知道airflow backfill到底是如何工作的，因为它可以帮助解释我在气流方面遇到的一些问题。它是否只是启动了某种调度器进程，而不考虑其余的DAG，并覆盖了start_date和end_date？或者不仅仅是这样？ 我的用例:有时我想为一个DAG重新运行一周，通常是当我向其添加一些新功能时。为此，我使用CLI (airflow clear)清除所需日期的任务实例。但有时，我可能会改变主意，认为这周不是重新运行的合适时机。因此，我暂停了DAG，并使用命令airflow backfill -m将清除的任务实例设置为成功。 当我这样做的时候，相关的任务实例会处于关于catchup=Tru
浏览 16提问于2019-06-22得票数 1
回答已采纳
3回答
使用python 2.7运行Airflow 1.9.0。如何优雅地停止DAG？
在这种情况下，我有一个DAG，它使用错误的代码运行文件上传，导致所有事情花费4倍的时间，我真的不希望等到它最终超时的一天(超时设置为10小时)。
DAG查找tar文件。当它找到一个文件时，它会遍历tar中的每个文件，查找要处理的文件，然后对它们进行处理。
我找不到任何方法来阻止DAG。我试着点击"DAG Runs“栏中的"Running”圆圈(右边的那个)。它允许我选择进程并将其标记为“失败”。但它并没有停止运行。
我试着点击“最近的任务”栏中的“运行”圆圈(左边的那个)。它允许我选择进程，但尝试将
浏览 1提问于2018-03-01得票数 7
1回答
停止气流回填运行
、、
我使用的是Airflow 1.9.0版本。
DAG每分钟都在运行。但是，当气流下降一段时间左右时，我们会看到大量回填作业正在运行，这会停止最近的DAG运行。
我使用的是SubDagOperator。
dag_args = {
'owner': 'Anish',
'depends_on_past': False,
'start_date': datetime.datetime(2017, 3, 20),#time in past
'retries': 0,
'retry
浏览 14提问于2021-06-12得票数 0
当我尝试在Airflow 1.8.0中运行DAG时，我发现完成前置任务的时间和后续任务执行的时间之间需要很长时间(通常比单个任务的执行时间要长)。Sequential、Local和Celery执行器的情况也是如此。有没有办法减少上面提到的开销时间？(就像airflow.cfg中可以加速DAG执行的任何参数一样？)新增甘特图以供参考：
浏览 16提问于2017-11-23得票数 22
所以，我对ETL管道很陌生，我正在尝试创建一个dag。进程很好地启动，第一个任务(unzip_data)运行，但其他任务无法运行。你能帮我指出原因吗？以下是完整的代码：
# importing required libraries
from datetime import timedelta
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.utils.dates import days_ago
#defining the DAG arguments
d
浏览 13提问于2022-04-20得票数 1
代码：
from datetime import datetime, timedelta
from airflow.operators.python import task
from airflow.operators.python_operator import PythonOperator
from airflow import DAG
@task
def get_content_body():
b = 1
print(b)
def get_content_body2(ti, **context):
a = 1
print(a)
defaul
浏览 4提问于2021-05-25得票数 1
回答已采纳
1回答
我搞不懂它是如何工作的气流运行两个任务并行。
这是我的达格
import datetime as dt
from airflow import DAG
import os
from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import PythonOperator, BranchPythonOperator
from airflow.contrib.sensors.file_sensor import FileSensor
from airflow.op
浏览 2提问于2020-03-09得票数 4
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
