有没有办法设计一个在dag之外实现特定数据管道模式的python类，以便将这个类用于所有需要该模式的数据管道？示例:为了将数据从Google云存储加载到Big Query，该过程可以通过数据质量测试来验证摄入候选文件。然后尝试在Big Query中加载原始表中的数据，然后根据加载结果将文件分派到存档或拒绝的文件夹中。 做一次很容易，如果需要做1000次呢？我正在试图弄清楚如何优化工程时间。任务组需要是dag的一部分才
浏览 39提问于2021-09-06得票数 1
回答已采纳
我有一个列表，比如说 import itertools task_id='cool', dag=dag) 最好的方法是编写一个循环吗？在我的例子中，主列表非常长，写出每个运算符将
浏览 13提问于2019-05-21得票数 0
所以我有一个任务A，它是将一些未知数量的文件复制到一个文件夹中。任务B在文件夹中的每个文件上运行。我没有办法预先知道文件的数量，因为它们一直在变化。有没有办法让它在气流中工作。gcpBucket', mime_type='text/plain',
dag=dag
浏览 3提问于2018-09-11得票数 2
有没有办法在Airflow中制作一个用户定义的宏，它本身是从其他宏计算出来的？from airflow import DAG 'simpletask = BashOperator( bash_command=
浏览 77提问于2017-07-01得票数 29
回答已采纳
2回答
import DAGERROR: test_execute (tests.operators.test_custom_operator.CustomOperatorTestCase)我没有办法调试。到目前为止
浏览 2提问于2019-07-04得票数 2
1回答
为了实现这一点，我将计时( HHMM )存储在Airflow变量中，并使用datetime.now()变量来获取当前的HHMM并创建以前运行的列表。但是，当Airflow dag每次都得到验证时，它会获取最新的日期和时间，并根据这些日期和时间生成新的先前任务列表。是否可以在运算符之外访问这些变量，因为上面的逻辑是创建要运行的动态任务列表，以检查之前的批处理运行完成情况。from pytz import timezone, utc
i
浏览 19提问于2020-05-08得票数 1
如果s3sensor操作员无法在s3位置找到文件，有没有办法跳过dag中的剩余任务。我知道我们可以使用短路运算符跳过这个任务，但我正在寻找一种方法，可以将短路运算符和s3sensor运算符集成在一起。如果没有，那么有没有其他方法可以做到这一点？提前谢谢你
浏览 23提问于2021-11-12得票数 0
3回答
我需要有几个相同的(仅在参数上不同的) DAG**s**，顶级也可以与以下约束/假设一起触发：
有没有其他/更好的方式将独立的(顶级)DAG连接<
浏览 2提问于2018-07-13得票数 16
回答已采纳
我试图将一个新的DAG添加到我们的Google Cloud Composer实例中-我们目前有32+ DAG-在中执行常规操作似乎没有任何效果-我们在we服务器/UI中看不到这些DAG，也看不到它们必须被加载我确实看到它们被复制到日志中适当的存储桶中，但除此之外什么也没有。最后，我组装了一个完全精简的DAG，并尝试添加它。下面是DAG：
fr
浏览 9提问于2020-03-11得票数 1
回答已采纳
有没有办法将命令行参数传递给Airflow BashOperator。目前，我有一个python脚本，它接受日期参数并执行一些特定的活动，比如清理比给定日期旧的特定文件夹。在只有一个任务的简化代码中，我想要做的是from airflow.operators import BashOperatorfrom airflow.models import DAG
from datet
浏览 5提问于2017-02-03得票数 10
在进入气流2.2之前，我们使用这个操作符触发另一个DAG和一个ExternalTaskSensor来等待它的完成。在气流2.2中，有一个名为wait_for_completion的新参数，如果设置为True，则只在触发的DAG完成时才使任务完成。
这很好，但我想知道这个工人是否会被释放。我知道ExternalTaskSensor过去有一个参数reschedule，您可以使用这个参数来处理大于1m的插槽，这将在pokes之间释放工作槽--但是我在文档中已经看不到了。我的问题是，wait_f
浏览 20提问于2022-04-06得票数 0
回答已采纳
2回答
有没有办法在任何运算符之外使用Airflow宏？例如，在DAG中，我有一个操作： bucket_name=bucketName,
dag</em
浏览 20提问于2017-07-13得票数 7
我创建了计算昨天日期的dag，并在所有dag的任务中使用它。有没有办法向dag传递一个参数(在我的例子中是一个日期)，并在我所有的操作符调用中使用它？(我不能改变操作员的代码，我只能使用它)，如果我的dag没有从外部获得任何日期-计算昨天的日期并在所有流程中使用它？所以基本上我想问我们是否可以用参数手动触发dag，如果dag没有得到任何参数来使用/计算默认值(在我的例子中是计算昨天的日期)
浏览 9提问于2019-08-06得票数 1
我在Google中使用Composer (气流)。我想要创建一个新环境，并将相同的DAG和变量从旧环境中移到新环境中。在我的新环境中，我导入了同一个JSON文件。
我使用gsutil并将相同的DAGs上传到新环境中。然而，在新的环境中，由于FERNET_KEY configuration is missing，我的所有DAG都崩溃了。我最好的猜测是，这与导入使用单独的Fernet密钥加密的变量有关，但我不确定。
浏览 5提问于2020-02-19得票数 6
回答已采纳
我正在使用Apache Airflow来管理数据处理管道。在流水线的中间，需要在下一步处理之前检查一些数据。我在考虑使用外部数据库来存储人工评审结果。并使用Sensor按时间间隔戳取审查结果。但在审查完成之前，它将占用一名气流工作人员。 有什么想法吗？
浏览 98提问于2018-02-02得票数 4
回答已采纳
我目前正在使用Airflow来运行一个有几个任务的DAG (比如dag.py)，然后，它有一个要执行的python脚本(通过bash_operator完成)。但我遇到的问题是，每当airflow扫描存储库的更改时(即每隔2分钟)，Airflow基本上都在运行这个python脚本report.py。有没有解决这个问题的办法？除了bash运算符之外，我们还能使用其他东西吗(请记住，在发送报告之前，我们需要做一些数据帧转换)？ 谢谢!
浏览 19提问于2021-01-06得票数 0
1回答
我想将这些依赖关系描述为DAG (有向无环图)，并在需要时执行该图。我一直在关注airflow，并写了一个虚拟脚本： from datetime import datetime
from airflow.operators.pythonpython_callable=cloud_runner)
first_task >> [second_task, second_task_bis] >> third_ta
浏览 64提问于2021-10-16得票数 0
在访问登录页面后，我使用res.cookie将jwt存储在cookie中，它在登录后正确地将jwt存储在cookie中。这是Postman中的cookie头参数。因此，在构建DAG定义文件时，我首先创建一个登录任务，以便其他操作员可以访问JWT并允许端点运行。t2和t3通过读取头和检查Cookie来要求jwt身份验证。import SimpleHttpOperator
浏览 0提问于2020-08-25得票数 2
3回答
在气流1.10.6中，决定一个任务是否应该重试的逻辑是在airflow.models.taskinstance.TaskInstance.handle_failure中，这样就不可能在操作员中定义行为，因为它是任务的责任，而不是操作员的责任。理想的情况是，如果handle_failure方法是在操作符的一侧定义的，那么我们可以根据需要重新定义它。我发现的唯一解决办法是使用PythonBranchingOperator“测试”任务是否可以运行。例如，在上述传感器的情况下，
浏览 2提问于2019-12-05得票数 5
回答已采纳
我正在尝试运行airflow作业，以使用MWAA(来自AWS的托管气流)在Snowflake实例上执行一些SQL操作。在MWAA中设置气流时，它要求DAG、插件和要求。有没有办法包含定制的sql脚本？下面是我尝试过的示例： with DAG( default_args=default_args,
schedule_interval为此，我将sql脚本"c
浏览 35提问于2021-08-09得票数 1
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
云直播活动推荐
腾讯云开发者
