airflow

1热度

1回答

您好我有我的DAG参数构造为使 default_args = { 'owner': 'airflow', 'depends_on_past': False, 'start_date': datetime(2017, 9, 26), 'schedule_interval': "* * * * *", 'email': ['[email protected]'], 'email_on_failure'

2热度

1回答

SQLAlchemy与空气流中的task_instance表的关系

我正在使用airflow，我希望能够跟踪表airflow.file_list中由给定任务实例生成的所有文件，这是与airflow（在postgres上运行）使用的相同数据库的一部分，。使用SQLAlchemy的我有我的file_list表下面的映射： from airflow.models import Base class MySourceFile(Base): """ SQLA

0热度

1回答

在重新启动的气流任务实例中恢复状态

我有一个Airflow操作员在第三方服务上踢任务，然后监视该作业的进度。在代码中，如果气流工人重新启动（通常是由于代码部署）的执行貌似 def execute(self, context): external_id = start_external_job() wait_until_external_job_completes(external_id) 这个任务的一个实例运

0热度

1回答

如何定义不应该定期运行的气流DAG /任务

目标非常简单：我需要为手动任务创建一个DAG，该任务不应该定期运行，而只能在管理员按下“运行”按钮时执行。理想情况下，无需切换“取消暂停”和“暂停”DAG（您知道某人肯定会忘记暂停）。到目前为止，我只有schedule_interval="0 0 30 2 *"（2月30日希望永远不会发生），但必须有更好的方法！有吗？

2热度

2回答

如何在EmailOperator任务的文件名中添加模板变量？（气流）

我似乎无法得到这个工作。我想每天发送一个给定的文件，其名称类似于'file _ {{ds_nodash}}。csv'。问题是，我似乎无法将此名称添加为文件名，因为它似乎无法使用。在电子邮件或主题的文本完美，而不是名称。这里是DAG作为一个例子： local_file = 'file-{{ds_nodash}}.csv' send_stats_csv = EmailOperator(

5热度

1回答

如何使用气流xcoms与MySqlOperator

def mysql_operator_test(): DEFAULT_DATE = datetime(2017, 10, 9) t = MySqlOperator( task_id='basic_mysql', sql="SELECT count(*) from table 1 where id>100;", mysql_conn_id='my

0热度

1回答

如何访问来自Airflow的响应SimpleHttpOperator GET请求

我正在学习Airflow并有一个简单的问题。下面是我的DAG称为dog_retriever import airflow from airflow import DAG from airflow.operators.http_operator import SimpleHttpOperator from airflow.operators.sensors import HttpSensor

1热度

1回答

如何删除XCOM对象，一旦DAG在Airflow中完成其运行

我在XCOM中有一个巨大的json文件，稍后我不需要一旦dag执行完成，但我仍然可以在UI中看到Xcom对象数据，一旦DAG运行完成后，是否有任何方式以编程方式删除XCOM？谢谢

0热度

1回答

Airflow会造成太多连接并导致MySQL RDS服务器崩溃

我正在使用Airflow连接来从RDS SQL实例中提取和处理数据。问题是该进程正在生成超过21个连接（尽管我的并发性在cfg文件中设置为4）并且RAM超载。根据我如何设置流程，我不希望每次连接超过6-8个连接。有没有一个好的cfg文件设置或管理面板选项，可以让我控制这种行为？我试过更新sql_alchemy_pool_recycle但这并没有解决问题。你可以看到系统性能开始坦克的GANT图表作

0热度

1回答

使用Airflow将数据从Postgres/MySQL移动到S3

我们正试图从Pentaho Kettle转移到Apache AIrflow来执行ETL，并将所有数据流程集中到1个工具下。我们使用水壶从Postgres/Mysql数据库每日读取数据，并将数据移至S3 - > Redshift。这样做最简单的方法是什么？我没有看到运营商可以直接做到这一点;所以我应该使用MySQL/Postgres运算符将数据放入本地文件中，并使用S3运算符将数据移至S3？谢