我希望将查询“从<table_name>中选择计数(*)”的结果从PostgresOperator中推送到PythonOperator，后者是pythonCallable，该值将用于python函数。我如何在其中使用XCom？我尝试过不同的方法，但每次我在python函数中得到一个未定义的变量时都会出错。def <python_function_name>():
""&
浏览 5提问于2022-07-13得票数 0
2回答
我试图在气流DAG中编写Python操作符，并将某些参数传递给Python可调用程序。 print(threshold)
task_id='poke_check', provide_context=True,
浏览 0提问于2019-02-15得票数 9
回答已采纳
我在Airflow DAG中使用了PythonOperator，并且我需要在操作符的Python函数中打印一些内容。我试着打印，但显然不起作用。我不太确定这是否会奏效。接下来，我尝试在PythonOperator中传递self.log，但我不确定如何传递该引用。task = PythonOperator( python_calla
浏览 10提问于2018-08-26得票数 5
回答已采纳
我在Python文件的文件夹中有一系列python任务: file1.py，file2.py，...
我读了Airflow文档，但我不明白如何在DAG中指定python文件的文件夹和文件名？我想要执行那些python文件(而不是通过Python操作符执行Python函数)。
浏览 19提问于2016-09-21得票数 29
回答已采纳
在Apache Airflow中，是否可以捕获失败的bash命令生成的原始错误消息，而不是Apache Airflow生成的跟踪错误，它会告诉您线路失败，但不能确切地说明失败的原因？Dag中的示例行： task_id="task1",
bash_command='gsutil rsync -r
浏览 2提问于2018-10-12得票数 2
我正在尝试将本地模块( python脚本)导入到我的DAG中。目录结构：├── dag│ └── my_DAG.py └── subfolder └── local_module.py#trying to import from local module
from script.subfolderimport local_
浏览 0提问于2018-05-03得票数 39
我正在创建我的第一个使用subdag的dag，并导入subdag运算符 `from airflow.operators.subdag import SubDagOperator` 然而，我一直收到流错误"Broken : /usr/local/airflow/dags/POC_Main_DAG.py没有名为‘airflow.operators.subdag’的模块“ 我还尝试导入虚拟操作符，但得到了相同的错误from airflow.operators.bash_
浏览 8提问于2021-01-06得票数 2
回答已采纳
我第一次尝试在我的笔记本电脑上配置Airflow (不使用docker，只遵循文档)。我的目标是设置一个简单的ETL作业。我已经用一个PythonOperator编写了最简单的DAG： from datetime import timedeltafrom airflowimport DAG
浏览 42提问于2020-11-08得票数 0
回答已采纳
我正在尝试使用docker中的airflow 2.0.1来运行我的python项目。下面是目录结构： dags在这里: path_to_airflow/dags/ python项目代码在这里: path_to_airflow/dags/utils 我遇到了处理异常的奇怪态度:我的PythonOperatorimport DummyOperator
from airflow.operators.python
浏览 25提问于2021-09-25得票数 0
回答已采纳
在我的Airflow DAG中，我有4 taskstask_4只在task_2和task_3成功运行之后才运行。如何设置条件，如：这是我的密码：from airflow.utils.datesimport days_ago
from
浏览 4提问于2020-10-20得票数 3
回答已采纳
我有一个带有调用函数的PythonOperator的DAG。我还有一个回调函数。我想把一些信息从可回调函数传递给回调函数。我该怎么做呢？我尝试使用上下文字典：from pprint import pprintfrom airflow.operators.python_operatorimport PythonOperator
浏览 9提问于2019-11-19得票数 1
我是Apache Airflow的新手。我的任务是从谷歌云存储读取数据，转换数据并将转换后的数据上传到BigQuery表中。我可以从云存储存储桶中获取数据，并直接存储到BigQuery表中。我不确定如何在这个管道中包含transform函数。下面是我的代码： # Import libraries needed for the operationfrom datetime import timedelta, datetim
浏览 18提问于2021-07-09得票数 1
回答已采纳
1回答
我在Windows中有以下dir/file结构： | |_dags
|_file_mover.py在first_dag.py中，我有一个触发file_mover.py的脚本。所以当我在终端做docker-compose up --build检查webserver localhost:8080时，我确实看到了气流中的first_dag。import DA
浏览 2提问于2021-06-16得票数 1
我第一次在GCP上工作，我试图使用bashoperator在计算引擎上执行python脚本。我将python脚本存储在composer脚本所在的桶中，但是当我试图运行bashoperator时，它会抛出一个错误文件，找不到。我能知道应该把python脚本放在哪里，以便在计算引擎上执行python脚本吗？bash_command='gcloud compute ssh --project '+PROJECT_ID+ ' --
浏览 1提问于2022-10-31得票数 0
回答已采纳
我在我的dag构建器模块中有一组任务，它使用Python操作符，就像全世界在Airflow中使用的那样。我正在kubernetes上使用docker部署airflow。pscript_pclean_zjg4qfamp9pda9jsxysyrqfj_AWFtK5ucowyw2
[2018-12-13 12:30:23,333] {bash_operator.py:97} INFO - Running command: python/usr/local/
浏览 15提问于2018-12-13得票数 2
from __future__ import print_functionfrom现在，我想将我的def print_context(ds, **kwargs)函数放到其他python文件中。因此，我创建了一个名为: simple_test.py和change的文件： task_id=&
浏览 4提问于2015-11-03得票数 25
( cloud composer的第一次用户)我见过的所有示例都在DAG中定义了非常简单的python函数。 我有多个冗长的python脚本要运行。我可以将这些放入任务中吗？如果是这样，那么是使用PythonOperator还是从BashOperator调用它们更好呢？schedule_interval = datetime.timedelta(days=1), default_args = default_dag_args) as dag:
do_stuff1 = pytho
浏览 17提问于2021-07-24得票数 2
回答已采纳
我被要求学习Docker，以便我们最终能够在Apache Airflow Docker + Kubernetes集群中设置我们的web抓取脚本。如果我用错了这些术语，那是因为我是码头的初学者。到目前为止，Apache Airflow确实有自己的映像，但是他们的表明，安装python包的唯一方法是通过pip install --user -r requirements.txt安装，我已经将pipinstall --user -r requirements.txt作为指令添加到自己的Dockerfile<em
浏览 23提问于2022-05-30得票数 6
我刚接触Airflow，只是为了练习在DAG中做一个简单的数据转换。然而，我不确定为什么第一个任务总是意外失败。有人能给我一些关于如何在DAG中调试失败的任务或指出错误部分的提示吗？非常感谢。import pandas as pdfrom airflow import DAG
from airfl
浏览 9提问于2021-09-13得票数 0
云服务器
ICP备案
实时音视频
对象存储
云直播
腾讯云开发者
