我第一次尝试在我的笔记本电脑上配置Airflow (不使用docker，只遵循文档)。我的目标是设置一个简单的ETL作业。import DAGfrom airflow.utils.dates import, )
run_etl 当我使用print语句传递一个伪函数时，<
浏览 42提问于2020-11-08得票数 0
回答已采纳
我想要实现的是创建一个任务，如果有的话，在那里发送通知--进程下的一个任务失败了。,dag=dag,)
batch1>>batch11
现在的问题是，当</em
浏览 0提问于2018-06-21得票数 2
我正在尝试运行一个气流DAG，需要为任务传递一些参数。
在python DAG文件中，如何读取在命令行trigger_dag命令中作为--conf参数传递的JSON字符串。例如：airflow trigger_dag 'dag_name' -r 'run_id' --conf '{"key":"value"}'
浏览 5提问于2017-08-30得票数 13
我使用Python调用天气API，但是气流任务由于错误Negsignal.SIGSEGV而失败。调用天气API的Python在气流外部运行时运行良好。DAGfrom airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operatorimport P
浏览 40提问于2022-09-02得票数 3
回答已采纳
2回答
想象一下一个简单的守护进程：from airflow.operators.python import PythonOperator if value: tags=['example']
run_this = <
浏览 4提问于2020-02-13得票数 0
回答已采纳
我正在使用Apache Airflow来管理数据处理管道。在流水线的中间，需要在下一步处理之前检查一些数据。，其中task1和task2是数据处理任务。task1完成后，task1生成的数据需要人工审核。在审查者批准数据之后，就可以启动task2了。人工审查任务可能需要很长时间(例如几周)。我在考虑使用外部数据库来存储人工评审结果。并使用Sensor按时间间隔戳取审查结果。但在审查完成之前，它将占用一名气流工作人员。 有什么想法吗？
浏览 98提问于2018-02-02得票数 4
回答已采纳
这些天，我正在做一个新的ETL项目，我想尝试一下作为工作经理的Airflow。我和我的同事都是第一次使用Airflow，我们采用了两种不同的方法:我决定编写python函数(类似apache-airflow项目中的运算符)，而我的同事使用airflow通过BashOperator调用外部python脚本。对我来说，主要的区别是：-使用BashOperator，你可以使用特定的python环境和特定的包调用python脚本-使用BashOperator，任务更加独立，并且
浏览 2提问于2017-11-28得票数 13
情况如下：from airflow import DAG<
浏览 12提问于2021-05-17得票数 0
1回答
我正在尝试建立一个ETL管道。第一步，我从api中获取一些数据。我希望这个进程在调用时立即运行。python dag.pyfrom airflow import DAG
浏览 0提问于2019-06-06得票数 1
5回答
气流失效松弛信息
、、
如何配置气流，使DAG中的任何故障(立即)导致消息松弛？", icon_url = 'http://airbnb.io/img/projects/airflow3.png',dag=dag)slack_failed_task <
浏览 5提问于2017-06-16得票数 5
回答已采纳
我使用Airflow Python操作符，它调用API从外部系统获取数据，解析数据并放入MongoDB (我想我也可以使用SimpleHttpOperator )。在被调用的方法中，我需要使用一个配置文件(director_api.cfg)，它具有外部系统的凭据以及Mongo实例的凭据。配置文件存储在gcp存储桶中。下面是dag代码：
from airflow.operators.python import
浏览 4提问于2022-10-04得票数 0
回答已采纳
1回答
在多次调用on_failure_callback功能时，我遇到了问题。目前，我的主要PythonOperator Dag被分成1 PythonOperator.和4 SubDag，每个SubDag由多个组成
t_python_1 >> t_subdag_2 >> t_subdag并检查任务时，似乎每个子Dag的任务都将on_failure_callback设置为None而不是None。因此，在执行期间，如果SubDag任务<em
浏览 9提问于2020-09-01得票数 0
回答已采纳
我创建了一个非常简单的DAG来使用PythonOperator执行Python文件。/dags puckel/docker-airflow webserver task_id='Tweets_load',
python_callable=njtransit_scrappe
浏览 1提问于2019-06-16得票数 5
我对气流很陌生，我正在尝试创建一个Python管道调度自动化过程。我的项目youtubecollection01利用自定义创建的模块，所以当我运行DAG时，ModuleNotFoundError: No module named 'Authentication'会失败这就是我的项目的结构：这是我的守护文件：# This to intialize the file as a dag file
from airflow import DAG</e
浏览 12提问于2022-07-31得票数 1
回答已采纳
我正在尝试创建一个气流管道，用于从API下载数据，处理数据，将其保存为CSV，然后将数据加载到Postgres数据库(所有数据都在码头容器中)。代码看起来像这样import pandas as pdfrom airflow.providers.postgres.ope
浏览 10提问于2022-01-27得票数 1
一切正常，main_dag运行得很好。它的PythonOperator函数正在被调用。但在Subdag中可调用的Python不会被调用。请帮帮我。import PythonOperatorfrom airflow.operators.dummy_operator=main_dag
AIRFLOW
浏览 0提问于2019-12-17得票数 2
我无法从Python连接到Airflow的元数据数据库。如果我尝试使用相同的连接，但从Python，它将不会工作。57315b5a登录: root但每当我创建一个DAG并尝试使用各种钩子从PythonOperator触发它时，我总是得到相同的错误消
浏览 0提问于2020-11-27得票数 1
我使用PostgreSQL作为数据库，CeleryExecutor和RabbitMQ作为芹菜后端。我创建了一些DAG，每个DAG都通过SSH隧道连接到Redshift数据库，并执行SQL命令。当我手动触发或通过调度程序运行时，每个DAG都会顺利运行。
但是，当我为这些DAG同时开始运行设置日程时，我遇到了一个错误。例如，如果DAG1和DAG2在上午8:00开始运行，这2个dags将失败并</
浏览 11提问于2021-01-12得票数 0
回答已采纳
我试图在PythonOperator、_etl_lasic之间将数据传递给另一个运行良好的PythonOperator _download_s3_data，但是当传递的值是None时，我想抛出一个异常，它应该将任务标记为失败。import airflowfrom airflow.operators.python import PythonOperator</e
浏览 9提问于2021-08-17得票数 0
回答已采纳
云服务器
ICP备案
云直播
腾讯会议
实时音视频
腾讯云开发者
