我在Airflow中编写了一个名为HadoopPutHdfs的自定义操作符，因此需要将xxx参数传递给HadoopPutHdfs，并使用generate_file_path任务的返回值填充xxx withhadoop_port=9870, dag=dag,
xxx= "{{ ti.xcom_pull(task_i
浏览 26提问于2021-06-24得票数 1
回答已采纳
1回答
我正在使用气流自定义操作符和来自任务id的xcom值(动态生成)。 def dummy_push_function(**context): url=
浏览 0提问于2022-04-11得票数 1
回答已采纳
我试图了解如何通过airflow xcom功能传递值。我试图构建的具体用途是编写一个文件，然后移动它，然后运行另一个命令。我的想法是将文件名从一个操作符传递到下一个操作符。airflow.operators.python_operator import PythonOperator dag_id='xcom_test_dag(key='filename', value=file_name)
def pull_function(**cont
浏览 0提问于2020-01-19得票数 1
回答已采纳
我想要GoogleCloudStoragePrefixSensor操作员的文件名。已尝试在主check_file_gcs所在位置使用context'ti'.xcom_pull(task_ids='check_file_gcs') prefix="<file_name>",
poke='10&#
浏览 3提问于2021-03-09得票数 1
回答已采纳
我是一个新的气流世界，我做了许多教程，将我的Python脚本之一转换为DAG。taskFour_operator = PythonOperator(task_id='taskFour', python_callable=taskFour, dag=dag)
我在任务中声明了以下依赖关系但是，显然(或者显然，与我的预期相反)，taskThree无法找到在taskOne和taskTwo中创建的映射数据集或数据集。在这方面，简单地将原始Python脚本分割成多个部分是行不通的。
浏览 1提问于2021-12-28得票数 1
回答已采纳
在这个解决方案中，我们还必须处理一个http端点的输出，并为下一个http enpoint生成输入。此外，根据触发器，可以同时调用相同的工作流。
浏览 28提问于2019-12-31得票数 2
1回答
我在一个组织中工作，它有大量令人尴尬的手工任务，涉及多个数据库(内部和外部)、多种文件类型/格式，以及大量的数据集。我试图在这里改进自动化，我一直在研究Luigi和Apache-气流，尝试标准化一些常用的模块，但我不确定这些是否是合适的工具。在我花太多时间弄清楚这些工具之前，我想我应该在这里问一下。从气流文档中也可以看到：“这是一个微妙但非常重要的问题:一般来说，如果两个操作符需要共享信息，比如文件名或少量数据，那么您应该考虑将它们合并成一个操作符。(尽管似乎
浏览 6提问于2022-08-23得票数 2
我已经在我的气流DAG中创建了一个任务，它返回一些值，这个值应该可以在xcom部分中看到，但是它不是。我设置了do_xcom_push = True，但它仍然没有显示出来。在此之前，我使用的是气流2.0.0，与将返回值推送给xcom的任务相同，但在气流2.1.2中，它没有发生。我不明白我错过了什么。下面是气流任务的一个片段：
任务= pytho
浏览 1提问于2021-10-19得票数 0
回答已采纳
1回答
我有一个进程，它使用CSV文件列表，然后将它们创建到导入的数据框架中。dag=dag)
但是如何在气流任务中定义它呢
浏览 2提问于2022-02-07得票数 1
回答已采纳
我们正在开发ETL工具，使用apache和apache气流。Apache气流将用于工作流管理。
能处理大量的数据吗？可以从apache气流中提取、转换计数吗？
浏览 1提问于2021-08-02得票数 0
我的用例是，我有一个S3事件，它触发一个lambda (在一个S3 createobject事件上)，它反过来调用一个气流DAG，传入两个-conf值(桶名，文件键)。然后，我将使用Python操作符提取键值，并将其存储在xcom变量中。然后，我希望在S3ToSnowflakeOperator中提取这个xcom值，并实质上将该文件加载到雪花表中。流程的所有部分都是工作栏，在S3ToSnowflake
浏览 1提问于2022-03-29得票数 1
2回答
我有一个气流DAG，它创建一个EMR集群，然后在该集群上运行SSHOperator任务。现在，我正在把EMR集群的主公共DNS硬编码成一个气流SSH连接。在创建EMR集群时，我的DAG是否可以动态填充这个DNS，这样我就不必手动更新连接了吗？
浏览 0提问于2019-05-02得票数 2
回答已采纳
这是我的操作员的代码： task_id="task1", location可以肯定地创建了sidecar容器，我在日志中看到了它的输出：Running command... [1mkill -s SIGINT 1[0m
INFO[
浏览 0提问于2019-10-11得票数 4
回答已采纳
我有一个存储的XCom值，我想传递给另一个python函数，它不是使用PythonOperator调用的。def sql_file_template(): task_instance = kwargs['task_instance']
task_instance.
浏览 5提问于2020-08-28得票数 2
我需要计算python运算符中的值，并在其他操作符中使用它，如下面所示，我得到了火花提交和电子邮件操作符/的"dag_var不存在“。
我将dag_var声明为python可调用的全局变量。
浏览 6提问于2022-04-12得票数 0
回答已采纳
我正在使用气流重试，所以我希望在失败后恢复相同的连接，并继续检查保存到xCom的xCom的状态。我能够将run_id推到xCom，但是当我为相同的任务提取它时，我得到的是None，而不是我在Admin仪表板中推送和看到的值(我可以将它拉到正在执行的其他任务中)。是否对从我试图从同一任务中推送的xCom推送值有任何限制？？我也可以选择使用气流变量而不是xCo
浏览 1提问于2021-04-25得票数 1
回答已采纳
我很高兴能定义气流中的工作流程。在下面的场景中，我无法获得xcom值(以黄色突出显示)。请查找示例代码下面的代码：def push_function(**context): provide_context=True,
url="http
浏览 0提问于2018-08-24得票数 4
回答已采纳
问题是，当我使用PostgresHook时，我得到了数据，但它在python方法中是我无法访问的，实际上我看到了下面的一行 task_instance.xcom_pushmy_value", value=details_json_string)
return d
浏览 3提问于2021-08-23得票数 3
回答已采纳
我一直在尝试在Airflow中使用PubSubPullSensor来拉取JSON消息并将其摄取到bigquery中，当消息大小超过阈值时，传感器无法通过XCOM拉取消息并推送。我知道XCOM有一个最大大小限制，但是我们如何克服这种情况，对于这种情况有什么变通办法吗？
浏览 3提问于2020-05-12得票数 0
我试着在气流中运行电子病历，并找到上面写的例子。step_adder = EmrAddStepsOperator( job_flow_id="{{ task_instance.xcom_pullstep_checker = EmrStepSensor( job_flow_id="{{ task_instance.<e
浏览 14提问于2022-04-03得票数 1
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
云直播
对象存储
腾讯会议活动推荐
腾讯云开发者
