变量在气流DAG中的应用
在气流DAG（Directed Acyclic Graph）中，变量的应用是为了在任务之间传递数据和状态。气流DAG是一种用于构建数据处理流程的工具，常用于数据工程和数据科学领域。
变量在气流DAG中的应用主要有以下几个方面：
- 数据传递：变量可以在不同的任务之间传递数据，实现数据的流动。例如，一个任务可以将处理结果存储在变量中，然后其他任务可以读取该变量并继续处理。
- 状态管理：变量可以用于管理任务的状态。在气流DAG中，任务的执行顺序是根据任务之间的依赖关系确定的。通过变量来记录任务的状态，可以确保任务在正确的顺序下执行。
- 参数配置：变量可以用于配置任务的参数。通过修改变量的值，可以改变任务的行为。这样可以灵活地调整任务的参数，以适应不同的需求。
- 错误处理：变量可以用于处理任务执行过程中的错误。例如，一个任务可能会在处理过程中发生错误，可以将错误信息存储在变量中，然后其他任务可以根据变量的值来进行相应的处理。
在气流DAG中，常用的变量类型包括全局变量和局部变量。全局变量可以在整个气流DAG中访问，而局部变量只能在特定的任务中访问。
腾讯云提供了一系列与气流DAG相关的产品和服务，包括：
- 腾讯云数据工厂（Data Factory）：提供了可视化的气流DAG编辑器，帮助用户快速构建和管理数据处理流程。
- 腾讯云批量计算（BatchCompute）：提供了高性能的批量计算服务，支持气流DAG作业的调度和执行。
- 腾讯云函数（Cloud Function）：提供了无服务器的计算服务，可以根据事件触发执行气流DAG作业。
- 腾讯云消息队列（Message Queue）：提供了可靠的消息传递服务，用于在气流DAG中传递数据和状态。
更多关于腾讯云相关产品和服务的详细介绍，请参考腾讯云官方网站：https://cloud.tencent.com/
相关·内容
我们可以参数化从气流变量中动态读取的气流schedule_interval，而不是直接作为cron表达式传递。
根据气流文档args ={ 'owner'：'pavan'，‘依赖于过去’：False，'start_date'：datetime(2020，1，15)，'email_on_failure'：True，'email_on_retry'：False，‘retry’：0，'on_failure_callback'：notify_email }
使用DAG(dag_id=DAG_NAME，def
浏览 3提问于2020-08-05得票数 4
回答已采纳
我用Openshift来部署气流。气流包括两个舱，气流网服务器和气流数据库(气流元数据库)。我已经创建了秘密，包括服务帐户的细节，如用户名和密码。我想访问数据管道中的那些值。
我的秘密部署配置
apiVersion: v1
kind: Secret
metadata:
name :airflow-service-account
namespace: CUSTOM_NAMESPACE
type: Opaque
stringData:
user-name : SERVICE_ACCOUNT_USERNAME_IN_BASE64
service-password: SERVICE_ACCOU
浏览 5提问于2022-02-16得票数 0
## Section 1 | Import Modules
## Section 2 | DAG Default Arguments
## Section 3 | Instantiate the DAG
## Section 4 | defining Utils
## Section 5 | Task defining
## Section 6 | Defining dependecies
## Section 1 | Import Modules
from airflow import DAG
from datetime import datetime
from airflow.op
浏览 3提问于2021-12-13得票数 1
回答已采纳
4回答
气流中的Python脚本调度
、、、、
大家好，
我需要使用气流调度我的python 文件(包含从sql和一些联接中提取的数据)。我已经成功地将气流安装到我的linux服务器上，而且我也可以使用气流的and服务器。但是，即使在阅读了文档之后，我也不清楚到底需要在哪里编写调度脚本，以及如何将该脚本应用到气流see服务器中，这样我就可以看到状态。
就配置而言，我知道dag文件夹位于我的主目录中的位置，也知道示例dags位于何处。
注意：请不要用如何在气流中运行bash脚本文件来标记这个副本，因为我需要运行位于不同位置的python。
请在气流网络服务器中找到配置如下：
下面是AIRFLOW_HOME dir中dag文件夹的屏幕
浏览 4提问于2017-01-18得票数 31
3回答
我想从气流UI中删除一个DAG，这在GCS/dags文件夹中不再可用。我知道气流有一种使用airflow delete_dag my_dag_id命令从DB中删除dags的“新”方法，这在中可以看到。
在composer气流版本中，似乎还不支持delete_dag命令。
不尝试这样做：--我也尝试过使用airflow resetdb --而且气流UI也死了
是否有方法删除当前不在gs://BUCKET/dags/文件夹中的dags？
浏览 1提问于2018-05-31得票数 5
回答已采纳
在这个中，我看到EmrCreateJobFlowOperator接收在气流UI中设置的aws/emr连接：
cluster_creator = EmrCreateJobFlowOperator(
task_id='create_job_flow',
job_flow_overrides=JOB_FLOW_OVERRIDES,
aws_conn_id='aws_default',
emr_conn_id='emr_conn_id',
dag=dag)
在气流UI中，在连接选项卡中，我如何添加AWS凭据，以便DAG能够获
浏览 1提问于2017-05-16得票数 2
回答已采纳
我对气流很陌生。我设置了一个气流环境，我的所有服务都是由气流触发的。但是，我打算直接从气流网将程序参数传递给DAG中定义的服务之一。我在气流网触发器上看到以下选项: DAG / conf
我想向DAG传递一个json，如下所示，作为我的程序参数：
{"list_key" : ["K1" : "V1", "K2" : "V2"]}
一旦我传递了上面所示的程序参数，我如何能够在我的DAG上访问这个参数？我的DAG python代码如下所示，但它引发了一个错误：
from airflow import DAG
浏览 5提问于2022-05-19得票数 1
是否有可能在气流中限制用户群对DAG的可见性和可访问性？
例如，我想有一个大的气流环境为我的整个公司，不同的团队将使用这个气流环境为他们的团队的工作流程。假设我们有A组和B组，他们都属于他们各自的AD/LDAP组、A组和B组。是否可以让A组只看到属于他们团队的DAG，反之亦然？
根据我的研究和理解，我认为在单一的气流环境下，这是不可能的。我想，为了让我做到这一点，我需要为每个团队创建一个单独的气流环境，以便每个团队都有自己的包含各自DAG的气流达格文件夹。
浏览 1提问于2018-06-27得票数 13
回答已采纳
1回答
我们正试图将每日(CSV)提取的文件合并到数据仓库中。
在我们的用例中，DAG的python代码对于我们所有的DAG都是相同的(~2000)，因此我们通过DAG生成器逻辑从一个文件生成它们。在我们的DAG中，我们只有15个任务(5个虚拟任务，2个CloudDataFusionStartPipelineOperator任务，8个python任务)。
在DAG生成过程中，我们读取气流变量(~30-50)来确定要生成哪些DAG(这也决定了DAG的ID以及它们应该处理的模式/表名称)。我们称之为发电机变量。
在DAG生成过程中，DAG还通过ID读取它们的配置(每个生成的DAG多出2-3个气流变量)。我
浏览 4提问于2022-04-10得票数 2
回答已采纳
1回答
如何为多个类别并行运行/实例化Dag(气流) ?
例如:我有一个气流( dag )，它是我定期运行的，我如何在不同的Batchname(Parallet)上调度dag并行运行：
运行batch1的进程(在args中传递批处理名称)
运行batch2的dag (在args中传递批处理名称)应该与1并行运行。。。
诸若此类
我使用环境变量传递Batchname，然后使用服务器上的多个tmux会话并行运行dag，但它被搞砸了。
有什么更好的方法可供我使用，并且可以节省时间并并行地运行多个批名称的dag？
耽误您时间，实在对不起。
浏览 0提问于2016-11-28得票数 1
2回答
我想运行一个像->这样的气流达格
我有两个气流工人W1和W2。
在W1中，我调度了单个任务(W1-1)，但是在W2中，我想创建X个任务数(W2-1，W2-2 ).W2-X)。
每个任务的数字X和bash命令将从DB调用中派生出来。
工作人员W2的所有任务都应该在W1完成后并行运行。
这是我的密码
dag = DAG('deploy_single', catchup=False, default_args=default_args, schedule_interval='16 15 * * *')
t1 = BashOperator
浏览 2提问于2018-07-02得票数 3
回答已采纳
假设我有一个简单的TaskFlow风格的DAG。
from datetime import datetime
from airflow.decorators import dag, task
from typing import Dict
@dag( start_date=datetime.now(),
schedule_interval='@once',
catchup=False)
def example_taskflow_api():
@task()
def extract(value=666) ->
浏览 16提问于2022-08-21得票数 0
我运行的气流从的正式码头-撰写。我想隐藏默认的DAG在网络应用程序。通常在airflow.cfg中可以这样做，如何将默认的DAG隐藏在坞-组合中？
干杯。
浏览 2提问于2021-10-02得票数 3
回答已采纳
我试图在气流中设置AWS过程，我需要EmrCreateJobFlowOperator中的EmrCreateJobFlowOperator和EmrAddStepsOperator中的steps通过位于其他地方的单独JSON文件来设置。
我尝试过许多方法，既可以直接链接JSON文件，也可以设置和获取JSON的气流变量。如果我要使用气流变量，它们也需要动态命名，这是我遇到的问题。我可以使用Variable.set轻松地使用PythonOperator实现动态名称，但由于气流在PythonOperator之外编写Python代码的局限性，无法在job_flow_overrides或steps中使用动态
浏览 1提问于2019-09-26得票数 2
2回答
我有一个气流DAG，它创建一个EMR集群，然后在该集群上运行SSHOperator任务。现在，我正在把EMR集群的主公共DNS硬编码成一个气流SSH连接。在创建EMR集群时，我的DAG是否可以动态填充这个DNS，这样我就不必手动更新连接了吗？
浏览 0提问于2019-05-02得票数 2
回答已采纳
1回答
我有一组气流DAG，每个都有一组与它们相关的标签。我还有一个PythonOperator，它被我所有的DAG使用。
如果DAG应用了特定的标记，则需要设置操作符的retries参数，而不是其他。
代码如下所示：
dag.py
from common import tasks
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from airflow.utils.helpers import chain
dag = DAG(
'test_retry_dag'
浏览 1提问于2021-11-19得票数 0
回答已采纳
1回答
我定义了一个全局变量timestamp，多个操作符引用这个变量。在运行每个操作符时，似乎会重新定义这个变量？下面是一个最小的可重复的例子。我预计test和test2都会打印相同的时间戳，但它们在气流中打印不同的时间戳(间隔几秒钟)。
import datetime
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
start_date = datetime.datetime(
year=2022,
month=3,
day=30,
hour=
浏览 6提问于2022-05-24得票数 0
我有一个Apache气流在Kubernetes上工作(通过Google )。我想在秘密中检索一个变量存储：
我需要使用气流中DAG中存储在这个秘密中的变量。(Python)
浏览 0提问于2018-10-05得票数 2
回答已采纳
1回答
我的气流中有一个的DAG脚本，它会自动刷新30 seconds.我希望禁用它(或者如果可能的话)设置为更高的时间限制.
此外，正如建议的一样，我已经将min_file_process_interval和dag_dir_list_interval设置为更高的值，并重新启动了气流服务器，但是在接下来的30秒内，DAG再次被刷新。
请建议一个解决办法来禁用此自动刷新或将其延迟到更长的时间间隔.谢谢
浏览 2提问于2020-11-25得票数 0
在气流网络接口中，我可以查看DAG的列表。如果单击其中一个DAG，我可以看到DAG的不同视图或DAG的代码，在details选项卡中，我可以看到DAGs文件的名称，但看不到路径。
我也尝试过airflow list-dags，但这也不能提供我想要的信息。
我可以在web接口中看到一个DAG，但是我找不到它在文件系统中的位置。
是否有方法从气流网络接口或气流CLI中找到DAG的位置？
浏览 8提问于2017-07-31得票数 0
回答已采纳
在我们的大数据项目中，有3000个表需要加载，所有这些表都应该由气流中的一个单独的DAG来处理。
在我们的解决方案中，单个python文件生成每种类型的表加载器，因此它们可以通过基于事件的REST通过云函数单独触发。因此，我们使用以下方法生成我们的DAG：
用于DAG生成器逻辑scd2的气流变量生成表类型的表名列表:插入追加、截断加载、scd1、
表加载器DAG的特定运算符使用的RawToRawRC_TableN 气流变量，例如:用于操作器处理的RR_TableN = {} // python = {} / python用于操作处理RawToCuration
用户定义宏:在任务定义之间，我们尝
浏览 3提问于2021-11-06得票数 1
回答已采纳
1回答
我已经在气流中为python中的函数编写了dag脚本，在dag脚本中，我导入了python脚本，作为"from data_extractor.py import *“，.I为data_extractor_dag.py.But中的data_extractor.py中的函数构建了一个工作流，当我在data_extractor.py中进行任何代码更改时，风流守护程序data_extractor_dag.py无法在这个场景中使用load.During，在这个场景中，气流UI页面也不是loaded.How以更好的方式使用气流。
浏览 0提问于2018-10-11得票数 1
对于Apache气流来说是新的，并且对代码和数据是如何在多节点气流设置中跨工作节点使用的很好奇。
当考虑ETL逻辑应该位于dag中的dag中，还是在dag中的操作符调用的单独文件中时，我从这里(https://www.astronomer.io/guides/dag-best-practices/)看到：
试着像配置文件一样对待DAG文件，并为钩子和操作符留下所有的重担。
由此，我的问题是：( 1)在多节点气流设置中，在工人之间同步操作员代码的最佳方法是什么？2)如果设置了多个节点，是否需要在所有工作人员之间安装共享驱动器以存储中间数据(或者气流是否有其他内部方式在工人之间传递数据以完成dag
浏览 0提问于2019-10-08得票数 1
回答已采纳
1回答
在处理多个环境(如Dev/Staging/Prod等)时，为每个环境设置单独(最好是相同的)气流实例将是最好的情况。
我正在使用GCP管理的气流()，这不是很便宜的运行，有多个实例将大大增加我们的每月账单。
那么，我想知道是否有人建议使用单个气流实例来处理多个环境？
我正在考虑的一种方法是在我的dags folder中有对应于每个环境的单独的顶级文件夹(即DAG /dev、dags/prod等)，并通过CI/CD管道将我的DAG脚本复制到相关文件夹中。
因此，在我的源代码存储库中，如果我的守护进程看起来像：
airflow_dags/my_dag_A.py
在CI阶段，我可以有一个构建步骤来创
浏览 6提问于2022-12-03得票数 0
我正在尝试运行一个气流DAG，需要为任务传递一些参数。
在python DAG文件中，如何读取在命令行trigger_dag命令中作为--conf参数传递的JSON字符串。
例如：airflow trigger_dag 'dag_name' -r 'run_id' --conf '{"key":"value"}'
浏览 5提问于2017-08-30得票数 13
我试图使用mssql+pyodbc连接字符串以Azure数据库作为后端运行气流(所有相关驱动程序都已安装)。
虽然气流能够连接到DB并创建表，即气流initdb成功运行，但我在运行气流调度程序时面临问题，因此触发的任务总是处于“运行”状态。
这是运行气流调度程序时遇到的错误：
*sqlalchemy.exc.ProgrammingError: (pyodbc.ProgrammingError) ('42000', "[42000] [Microsoft][ODBC Driver 17 for SQL Server][SQL Server]Incorrect syntax
浏览 4提问于2020-12-16得票数 0
1回答
在触发DAG之前，是否有一种快速/简单的方法来检查气流(1.10) DAG脚本中的语法错误、任务错误等。
在我们的项目中，我们必须首先将DAG代码推到存储区，然后通过Jenkins部署它，然后才能真正运行DAG。因此，只有在触发错误后才能发现错误。
当我们不得不反复通过相同的过程来发现和修复每一个小的编码错误时，这就变得非常乏味了。
因此，在触发DAG之前，检查是否有快速检查和修复这些错误的方法。
气流版本- 1.10
谢谢
浏览 5提问于2022-09-02得票数 2
我们使用的是气流版本2，当我们将DAG文件放置在DAG文件夹中时，需要20到30秒才能在气流数据列表中反映这个dag。
注意事项：我们使用芹菜执行器和postgres作为数据库。
有人能帮我如何使气流更快地拿起DAG吗？它是否有一个配置/任何想法都是值得注意的。
谢谢你，哈利
浏览 3提问于2021-09-19得票数 1
回答已采纳
1回答
使用BashOperator从气流DAG任务中触发ansible剧本，因为我想获取在task 1中生成的值，并使用那些在气流DAG任务2中生成的值。
任务1执行生成此值的ansible剧本。在气流DAG任务2中是否有获得这个值的方法?遵循下面的doc，但是找不到涵盖ansible剧本用例的任何示例代码。
浏览 3提问于2022-09-23得票数 0
2回答
我正试图实现一种方法来获取气流变量中的动态值。喜欢
只想问一问，是否有任何方法可以在运行时插入DAG_NAME和DateTime.now值，这是在DAG文件中定义的，因此最终的结果将是这样的：“在9月22日-2021年12:00:00启动0_dag_1”。
浏览 5提问于2021-09-22得票数 1
我们目前正在使用气流在运行时部署流，并在DAG部署方面面临不断的问题。在我们的工作流应用程序中，DAG不会被Scheduler及时选择，并且会延迟用户响应。
是他们的任何方式，我们可以将DAGs部署到按需气流数据库中。
？如果不是，我们如何才能使这一过程得到明确的定义。
浏览 1提问于2021-04-20得票数 0
我们有一个前端服务器，它直接使用dag执行dags (DagBag()，get_dag()，然后是dag_run())。
Dags运行良好，问题是，我们无法找到一种使用特定参数来执行此类dags的方法。
最接近的解决方案是使用变量API，它使用set()和get()方法，但是这些变量是全局的，在使用相同变量名的并发操作时可能存在冲突。
我们如何运行一个dag并设置可用于其执行的参数？我们主要是使用PythonOperator。
编辑1:我们的程序是一个Python前端服务器。因此，我们通过另一个Python程序来讨论气流。这意味着我们通过Python触发dags，因此，我们使用DagBag.g
浏览 4提问于2017-07-14得票数 0
1回答
我正在研究气流DAG。我有两个DAG，分别是DAG A和DAG B。我想在DAG B中基于DAG A的execution_date做一些事情。为此，我使用了气流变量。 在DAG A中： def set_execution_date(**kwargs):
Variable.set('var_name',kwargs['execution_date'])
status = Variable.get(time_status) 这将使用DAG A的execution_date更新我的气流变量，如下所示： ? 现在，我使用DAG B中的airfl
浏览 20提问于2021-06-17得票数 1
回答已采纳
我正在学习Apache气流，并尝试编写hello world版本的它。
我已经用python版本3.8和下面的包含dag和任务的代码安装了airflow版本2.0
import datetime
import logging
from airflow import DAG
from airflow.operators.python import PythonOperator
def greet_hello():
logging.info("Hello World")
dag = DAG("FirstDag", start_date=dateti
浏览 7提问于2021-02-07得票数 2
回答已采纳
2回答
我试图打印在气流数据库中计算出来的变量的内容，因此我在bash操作符中使用了echo，但是它不起作用
我尝试使用一个预定义的变量，但得到了相同的输出。
以下是一个例子：
with DAG(
dag_id="tmp",
) as dag:
test="test"
test_dag = BashOperator(
task_id="test_dag",
bash_command='echo $test',
)
输出始终为空：
输出：
INFO -
浏览 2提问于2022-06-14得票数 1
回答已采纳
请参阅：
。
考虑到这个概念，并试图应用于Apache气流。我在Windows Linux子系统(WSL)上运行气流。我还在Windows10上安装了和.py文件，我想使用Airflow来执行.py文件。
如果我在Ubuntu中执行以下操作：
/mnt/c/path/to/windows_python/python.exe C:\\path\\to\\my\\python\\file\\my_file.py
my_file.py使用我安装的Python 10运行。但是，如果在DAG中使用BashOperator进行相同的调用，则会看到"Bash命令失败“错误。DAG文件中的任务如下所示
浏览 0提问于2019-08-14得票数 0
回答已采纳
1回答
我有以下的文件夹结构在气流达格文件夹。
/opt/airflow/dags
--my_module/
--my_dag.py
Dag文件(my_dag.py)需要my_module中的python脚本来执行工作流任务。
上面的设置运行得很完美，但是当更新"my_module“文件夹中的python脚本时，它不会反映在工作流中。
当更改dag文件时，它将反映气流中的变化。我也重新启动了气流，但没有运气。我在这里做错了什么..。
浏览 0提问于2018-08-28得票数 0
回答已采纳
2回答
我们有一个连续的集成管道，自动将我们的气流DAG部署到气流服务器。在部署新版本的DAG时，默认情况下其状态为OFF。我们希望将其转换为ON，作为部署过程执行的任务的一部分。
气流中是否有允许打开DAG的命令行选项？谢谢
浏览 3提问于2017-11-22得票数 6
回答已采纳
5回答
我想弄清楚气流是否支持跳过DAG中的任务来执行临时执行？
假设我的DAG图如下所示: task1 > task2 > task3 > task4
我想从task3手动启动我的DAG，最好的方法是什么？
我读过关于ShortCircuitOperator的文章，但我正在寻找更多的即席解决方案，一旦执行被触发，就可以应用它。
谢谢!
浏览 1提问于2018-09-05得票数 12
回答已采纳
1回答
我使用的气流2.2.2，并有一个dag，这是计划运行每10分钟，并暂停。我正在尝试使用气流客户端手动调用它。Dag没有被取消暂停，dagrun也在排队，声明。在创建dag运行时，是否可以在不调用其他API调用的情况下使用气流客户端来取消dag。
api_instance = dag_run_api.DAGRunApi(api_client)
dag_run = DAGRun(
logical_date=datetime.now(timezone(timedelta())),
conf=request_data,
浏览 6提问于2022-05-16得票数 1
我一直在学习气流，并为ETL管道编写DAG。它涉及到使用AWS环境(S3，Redshift)。它处理将数据从一个桶复制到另一个桶，然后将其存储在红移中。我将桶名和前缀作为变量存储在气流中，为此您必须打开GUI并手动添加它们。
这是业界最安全和最广泛使用的做法，除下列选项外
我们可以使用airflow.cfg来存储变量(桶名)并在DAG中访问它们吗？
使用自定义配置文件并使用configparser解析其内容
使用GUI添加变量
浏览 1提问于2019-08-12得票数 0
回答已采纳
1回答
我有一个spark-streaming作业，运行在EMR上，由气流排定。我们每周都要优雅地终止这个电子病历集群。
但是，当我向运行中的kill应用程序发出SIGTERM或spark-streaming信号时，它在气流DAG中报告为“失败”任务。这将阻止DAG进一步移动，防止下一次运行触发。
有什么方法可以杀死正在运行的spark-streaming应用程序来标记成功，或者让DAG完成，即使它认为任务失败了？
浏览 1提问于2018-07-23得票数 0
1回答
我是在气流中的新人，但我不知道如何正确地发布新的python脚本或jars到气流服务器。问题的部分是/home/airflow/airflow/dags/scripts/file1.py.如果我创建了一个新版本的file1.py，我想在气流服务器上传播这个新版本。简单的解决方案可以是通过FTP
我找到了，这可能是我在应用程序中的第一步，并获得了我想要的脚本版本。
( 1)在Aiflow，是否有比单独一步更好的方法？
2)问题的第二部分是如何处理jar文件。我的解决方案是在某个地方发布jar文件(例如，Nexus存储库)，并在气流中有一个单独的步骤来下载我的文件的最新版本。
from airfl
浏览 2提问于2020-03-03得票数 0
我正在一个单独的虚拟env中运行气流，并运行一些具有特定需求的数据质量DAG。实际上，我希望在他们自己的虚拟env中运行DAG，而不是杂乱的基础气流环境。
pythonVirtualenvOperator做了一些类似的工作，但它每次都会创建自己的环境，然后再删除它们。一天运行几次的DAG在时间或空间上都没有效率。
在相同的气流安装中，我无法在单独的虚拟环境中运行DAG。有什么办法吗？
浏览 5提问于2022-03-24得票数 0
1回答
我正在评估气流是否适合我的需要(在生物信息学中)。我对气流模型有些困难。具体地说：
DAG文件实际上在哪里执行？它的背景是什么？如何将输入数据传递到DAG定义文件中？(例如，我希望为目录中的每个文件创建一个任务。)
如何临时执行DAG？如何传递DAG结构的参数？
下面是我想要执行的一个例子。假设我刚收到一些数据作为一个目录，其中包含一些共享文件系统中的20个文件。我想要执行一个DAG管道，它在20个文件中的每个文件上运行一个特定的bash命令，然后组合一些结果并执行进一步的处理。DAG需要文件系统上的路径，还需要列出目录中的文件，以便为每个文件构造一个任务。
我可能没有必要将元
浏览 4提问于2017-05-26得票数 1
2回答
我试图使用气流在BigQuery中创建一个参数化查询。将在气流中设置的参数，并将更改在BigQuery中运行的查询中的值。
例如，这是我的查询：
从表中选择id ={{params.number_days}
气流代码包括：
( destination_dataset_table='prod.table_result'，_id=‘table’，use_legacy_sql=False，WRITE_dag=dag=‘WRITE_TRUNCATE’，allow_large_results=True，bql=merchant_rank_query，dag=dag= {'
浏览 5提问于2020-12-02得票数 2
2回答
我们正在构建工作流调度应用程序。我们发现气流是工作流管理器的好选择，Kubernetes是集群管理器的好选择。因此，流动将是，
我们将提交工作流DAG到气流。
气流应通过指定码头图像将给定DAG的任务提交给Kubernetes。
Kubernetes应该通过在集群的可用EC2工作节点上运行docker容器来执行任务。
在搜索过程中，我们发现，气流有与ECS、Mesos集成的操作符，但对于Kubernetes没有。但是，我们在上找到了对Kubernetes操作符的请求，但没有对其进行任何进一步的更新。
所以，简单地说，问题是，如何将气流与库伯奈特结合起来？
浏览 0提问于2018-01-24得票数 12
回答已采纳
我的应用程序设计要求一个气流DAG，它目前正在执行，可以暂停，如果执行，恢复，如果暂停和停止。这与提供的暂停/取消暂停功能不同。例如，如果DAG有8个组件，3个组件已经成功执行。现在我想暂停这个DAG的执行。过了一段时间，我想恢复暂停的DAG。在完成了6个组件之后，我意识到这个DAG有一个很大的缺陷，我需要终止它的执行。
我试图在中查找这三个功能，但是找不到这些函数。也没有告诉任何能够做到这一点的API。
会要求气流退伍军人对此提出建议。
浏览 0提问于2019-04-16得票数 2
全,
我们有几个带有DAG信息的JSON文件。我创建了一个python脚本来读取那些JSON文件，以便为Composer气流动态生成DAG。现在，我面临的挑战是如何在python脚本中获得运行DAG的列表，并从本地系统触发单个DAG。
在我的本地，我设置了私有SSH键，所以我可以使用下面的命令获得在composer气流中运行的DAG列表。
gcloud -q composer environments run "$ENVIRONMENT" --location "$REGION" list_dags
我可以通过使用气流服务器URL来使用POST/GET请求，但是
浏览 2提问于2019-12-18得票数 0
回答已采纳
我正在致力于一个功能，需要schedule_intervals的气流作业。我没有自己编写代码来解析DAG文件中的cron表达式，而是尝试在气流元数据DB中找到解析的schedule_interval值，但没有结果。
有人能给我一个指针，说明气流如何解析schedule_interval表达式(例如，中的一个文件)，以及它将解析的值存储在哪里(如果存储这些值)？
编辑：
上面的schedule_interval表达式是DAG参数schedule_interval，如: dag = DAG(“教程”，default_args=default_args，schedule_interval=‘@D
浏览 1提问于2018-07-24得票数 1
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
