使用S3ToRedShitOperator执行截断红移表时出错
我想在将CSV文件加载到RedShift表之前截断RedShift表。 错误： airflow.exceptions.AirflowException：传递给S3ToRedshitOperator(task_id：dag_run_s3_to_redshift)的参数无效。无效参数为： **kwargs：{‘method’：‘place’} 以下代码： task_fail_s3_t
..
airflow-2.x相关内容
升级气流版本后，远程日志记录不起作用
我们将Airflow v2.1.4与Kubernetes执行器一起使用。 我们的K8S群集Pod stdout通过filebatch->；logstash->；Elasticsearch(Elk)自动发货。 在logstash中，我们正在创建log_id字段： mutate { copy => { "[log][offset]" => "offset" }
..
Airflow 2.0.1:在本地时区写入日志
默认情况下，airflow 2.0.1 在 UTC 时区写入日志.但我希望日志以本地/机器 [非 UTC] 时区写入.我尝试在下面更改.但没有运气 AIRFLOW__WEBSERVER__DEFAULT_UI_TIMEZONE:亚洲/加尔各答 AIRFLOW__CORE__DEFAULT_TIMEZONE:亚洲/加尔各答 解决方案 您需要在 cointainers 中设置时区.
..
Airflow 2.0.2 如何使用 xcom 在 postgres 任务中传递参数?
我正在尝试以动态方式在 postgres 运算符中传递参数. 刷新元数据有两个任务， 获取id列表(get_query_id_task) 传递 id 列表以获取并执行查询 (get_query_text_task) get_query_id_task = PythonOperator(task_id='get_query_id',python_callable=query_an
..
使用 XCOM 值在 Airflow 中创建动态工作流
现在，我使用这样的变量创建了多个任务，并且工作正常. 使用 DAG(....) 作为 dag:body = Variable.get("config_table", deserialize_json=True)对于范围内的 i(len(body.keys())):simple_task = 操作员(task_id = 'task_' + str(i),..... 但出于某种原因我需要使用 X
..
根据气流中的 sql 查询结果创建动态任务
我正在尝试使用 TaskGroup 创建动态任务，并将结果保存在变量中.根据数据库查询，该变量每 N 分钟修改一次，但是当第二次修改该变量时，调度程序会崩溃 基本上我需要根据查询中收到的唯一行数创建任务. 以 TaskGroup(f"task") 作为任务: data_variable = Variable.get(“df")数据 = 数据变量尝试 :如果数据 != False 和
..
Docker编写气流2(版本2.0.0)的文件
我希望编写docker compose文件，以在类似环境的生产环境中本地执行气流. 对于较老的气流v1.10.14，泊坞窗compose运行良好.但是，相同的docker compose不适用于最新的稳定版本，airflow Scheduler&Web服务连续失败.错误消息看起来像无法创建审核表. docker-compose.yaml : 版本:"2.1"服务:postgres:图
..