Python
Java
PHP
IOS
Android
Nodejs
JavaScript
Html5
Windows
Ubuntu
Linux
Airflow:为每个文件运行 DAG 的正确方法
我有以下任务需要解决 文件通过端点不定期发送并存储在本地 我需要为每个文件触发 DAG 运行 对于每个文件 将执行相同的任务 总体流程如下 对于每个文件 运行任务 A gt B gt C gt D 正在批量处理文件 虽然这项任务对我来说似乎
python
ETL
airflow
batchprocessing
directedacyclicgraphs
是否可以访问 AWS MWAA 中的 Airflow API?
我正在运行一个 AWS MWAA Airflow v2 0 2 实例 我想访问气流API 目前支持这个吗 计划未来发布 AWS 文档中没有提及这一点 看起来可以在 AWS MWAA Airflow v1 10 12 中启用实验性 API 但
amazonwebservices
airflow
mwaa
Airflow 任务失败/重试工作流程
我有任务的重试逻辑 但不清楚重试打开时 Airflow 如何处理任务失败 Their 文档只是指出on failure callback当任务失败时被触发 但是如果该任务失败并且也被标记为重试 这是否意味着on failure callba
airflow
Airflow:如何在非PythonOperator中使用xcom_push和xcom_pull
我看到很多关于如何使用的例子xcom push and xcom pull与 Airflow 中的 PythonOperators 一起使用 我需要去做xcom pull from a 非Python运算符类 但找不到如何做 任何指针或示例
airflow
Airflow 身份验证设置失败并显示“AttributeError:无法设置属性”
Airflow 版本 1 8 密码身份验证设置 如中所述docs在该步骤失败 user password set the password 有错误 AttributeError can t set attribute 最好简单地使用Pass
airflow
Airflow:为什么操作员有一个开始日期?
我不明白为什么我们需要操作员 任务实例 的 开始日期 我们传递给 DAG 的那个不就足够了吗 另外 如果当前时间是 UTC 2018 年 2 月 7 日上午 8 点 30 分 现在我将 dag 的 start date 设置为 2018 年
airflow
airflowscheduler
airflow TriggerDagRunOperator 如何更改执行日期
我注意到 对于计划任务 执行日期是根据过去设置的 Airflow 是作为 ETL 需求的解决方案而开发的 在 ETL 世界中 您通常会汇总数据 所以 如果我想总结数据 2016 02 19 我会在格林威治标准时间2016 02 20午夜进行
triggers
airflow
将字符串列表作为 Airflow 中相关任务的参数传递
我正在尝试通过以下方式将字符串列表从一个任务传递到另一个任务XCom但我似乎无法将推送列表解释回列表 例如 当我在某些函数中执行此操作时blah这是运行在ShortCircuitOperator paths gs format bucket
python
airflow
无法使用 SparkSubmitOperator 执行 Spark 作业
我能跑Spark工作使用BashOperator但我想用SparkSubmitOperator对于它使用Spark 独立模式 Here s my DAG for SparkSubmitOperator and 堆栈跟踪 args owner
airflow
气流 外部传感器卡在戳戳处
我希望一个 dag 在另一 dag 完成后开始 一种解决方案是使用外部传感器功能 您可以在下面找到我的解决方案 我遇到的问题是依赖的 dag 卡在 poking 上 我检查了这个answer并确保两个 dags 按相同的时间表运行 我的简化
airflow
airflowscheduler
MWAA - 气流 - PythonVirtualenvOperator 需要 virtualenv
我正在使用AWS的MWAA服务 2 2 2 运行各种 DAG 其中大多数都是使用标准 PythonOperator 类型实现的 我将 DAG 与任何共享需求一起捆绑到 S3 存储桶中 然后将 MWAA 指向相关对象和版本 到目前为止一切都进
python
amazonwebservices
airflow
virtualenv
mwaa
如何在 Airflow 中运行异步函数?
我正在编写一个气流任务来读取大型 csv 并将其保存到 postgresql 数据库 我发现这个 asyncpg 包具有复制功能 其运行速度比任何其他包都要快得多 然而 它是异步的 我不知道如何将它合并到Airflow中 这是示例代码 fr
python
asyncawait
airflow
运行 tweepy 的 Airflow 任务退出并返回代码 -6
我有一个简单的 Airflow DAG 它只有一个任务 stream from twitter to kafka 以下是 DAG 的代码 default args owner me depends on past False start d
python
airflow
tweepy
Airflow 在 1 分钟后终止了我的任务
我有一个非常简单的 DAG 有两个任务 如下所示 default args owner me start date dt datetime today retries 0 retry delay dt timedelta minutes 1
python
directedacyclicgraphs
airflow
如何从通过 SSHExecuteOperator 推送的 Airflow XCom 检索值
我有以下 DAG 和两个 SSHExecuteOperator 任务 第一个任务执行一个返回参数的存储过程 第二个任务需要此参数作为输入 请解释一下如何从任务 1 中推送的 XCom 中提取值 以便在任务 2 中使用它 from airfl
SSH
airflow
如何跳过Airflow操作员中的任务?
Airflow 有没有办法从 PythonOperator 中跳过当前任务 例如 def execute if condition skip current task task PythonOperator task id task pyt
python
airflowscheduler
airflow
我们如何使用 TriggerDagRunOperator 触发多个气流 dag?
我有一个场景 其中特定的 dag 完成后需要触发多个 dag 已使用 TriggerDagRunOperator 触发单个 dag 是否可以将多个 dag 传递给 TriggerDagRunOperator 来触发多个 dags 是否可以仅
directedacyclicgraphs
airflow
apache-airflow 1.9 默认时区设置为非 utc
我最近将气流版本从 Airflow 1 8 升级到 apache airflow 1 9 升级成功 并且我使用 Celery Executor 扩展了环境 一切似乎都工作正常 但 dag 和任务开始日期 执行日期等都出现在UTC 时区和计划
timezone
UTC
airflow
如何解析气流模板中的json字符串
是否可以在气流模板中解析 JSON 字符串 我有一个 HttpSensor 它通过 REST API 监视作业 但作业 ID 位于上游任务的响应中 该任务具有xcom push marked True 我想做类似下面的事情 但是 这段代码给
airflow
Airflow Trigger DAG with config
1 rest api to trigger dag POST api experimental dags
工具及软件
airflow
DAG
«
1 ...
8
9
10
11
12
13
14
»