27
4
I have a table in my DB where I'm logging some statuses.
If the status is ok for some date I should run sending_reports_if_success task, if it's not ok then sending_reports_if_failed.
我在数据库中有一个表，记录一些状态。如果某个日期的状态正常，我应该运行sending_reports_If_success任务，如果不正常，则sending_reports_If_failed。
How can I create a making choice task which will get result from
sql :
如何创建一个从sql中获得结果的决策任务：
个人中心
文章发布
退出select count(*) from r_summary rs
where rs.status = 'Not Ok'
and rs.created_date = '{start_task_time}';
And if count = 1 then I should run failure task, if count = 0 then sucess task
如果count=1，那么我应该运行失败任务，如果count=0，那么成功任务
BranchPythonOperator is suitable for you case. You can pass the result from your query task to branching task and it will direct the flow to correspondingly using task ID.
BranchPythonOperator适合您的情况。您可以将查询任务的结果传递给分支任务，它将使用任务ID相应地引导流。
from airflow.operators.python import BranchPythonOperator
def choose_branch(query_result):
if query_result == 1:
return "sending_reports_if_success_task_id"
return "sending_reports_if_failed_task_id"
branching = BranchPythonOperator(
task_id='choose_report_branch',
python_callable=choose_branch,
op_args=["{{ ti.xcom_pull('your_query_task', 'your_return_key') }}"]
)
Thanks, but how can write a task that will return a key? def extract_data(conn_id, **context): engine = get_conn(conn_id) df = pd.read_sql_query(f'''select * from r_summary rs where rs.status = 'OK' and rs.created_date = '{start_task_time}'::timestamp with time zone;''',con=engine) return df checking_exists_reports = PythonOperator( task_id='checking_exists_reports', python_callable=extract_data, op_kwargs={ 'load_date': '{{ts}}', 'conn_id' : 'postrgres' } )
谢谢，但是如何编写一个返回密钥的任务呢？def extract_data（conn_id，**context）：engine=get_conn（conn_id）df=pd.read_sql_query（f''从r_summary rs中选择*，其中rs.status=“OK”和rs.created_date=“｛start_task_time｝”：：带时区的时间戳；''，con=engine）返回df-checking_exists_reports=python操作员（task_id=“checking_exists-reports”，python_callable=extract_date，op_kwargs=“load_date”：“｛｛ts｝}'，'conn_id'：'postrgres'｝）
you can use context["task_instance"].xcom_push(key="your_query_task", value="your_value") in extract_data function to push desired value to xcom under your_query_task key.
您可以在extract_data函数中使用context[“task_instance”].xcom_push（key=“your_query_task”，value=“your_value”）将所需值推送到your_query.task键下的xcom。
27
4
0
从 Redis 获取消息时，onDone:(){print('done')} 从未起作用。 import 'package:dartis/dartis.dart' as redis show PubS
昨天我玩了一些vim脚本，并设法通过循环来对当前输入的内容进行状态栏预测(请参见屏幕截图(灰色+黄色栏))。 问题是，我不记得我是怎么得到的，也找不到我用于该vim魔术的代码片段(我记得它很简单):它
我尝试加载 bash_completion在我的 bash (3.2.25) 中，它不起作用。没有消息等。我在我的 .bashrc 中使用了以下内容 if [ -f ~/.bash_completio
我正在尝试构建一个 bash 完成例程，它将建议命令行标志和合适的标志值。例如在下面 fstcompose 命令我想比赛套路先建议 compose_filter= 标志，然后建议来自 [alt_seq
当我尝试在重定向符号后完成路径时，bash 完成的行为就好像它仍在尝试在重定向之前完成命令的参数一样。 例如: dpkg -l > /med标签 通过在 /med 之后点击 Tab我希望它完成通往 /
我的类中有几个 CAKeyframeAnimation 对象。 他们都以 self 为代表。 在我的animationDidStop函数中，我如何知道调用来自哪里？ 是否有任何变量可以传递给 CAKe
我有一个带有 NSDateFormatter 的 NSTextField。格式化程序接受“mm/dd/yy”。 可以自动补全日期吗？因此，用户可以输入“mm”，格式化程序将完成当前月份和年份。 最佳答
有一个解决方案可以使用以下方法完成 NSTextField : - (NSArray *)control:(NSControl *)control textView:(NSTextView *)tex
我正在阅读 Passport 的文档，我注意到 serialize()和 deserialize() done()被调用而不被返回。 但是，当使用 passport.use() 设置新策略时在回调函数
在 ubuntu 11.10 上的 Firefox 8.0 中，尽管 img.complete 为 false，但仍会调用 onload 函数 draw。我设法用 setTimeout hack 解决
假设我有两个与两个并行执行的计算相对应的 future 。我如何等到第一个 future 准备好？理想情况下，我正在寻找类似于Python asyncio's wait且参数为return_when=
我正在寻找一种 Java 7 数据结构，其行为类似于 java.util.Queue，并且还具有“最终项目已被删除”的概念。 例如，应可以表达如下概念: while(!endingQueue.isFi
这是一个简单的问题。 if ($('.dataTablePageList')) { 我想做的是执行一个 if 语句，该语句表示如果具有 dataTablesPageList 类的对象也具有 menu
我用replaceWith批量替换了许多div中的html。替换后，我使用 jTruncate 来截断文本。然而它不起作用，因为在执行时，replaceWith 还没有完成。 我尝试了回调技巧 ( H
有没有办法调用 javascript 表单 submit() 函数或 JQuery $.submit() 函数并确保它完成提交过程？具体来说，在一个表单中，我试图在一个 IFrame 中提交一个表单。
我有以下方法: function animatePortfolio(fadeElement) { fadeElement.children('article').each(function(i
我刚刚开始使用 AndEngine， 我正在像这样移动 Sprite : if(pValueY < 0 && !jumping) { jumping =
我正在使用 asynctask 来执行冗长的操作，例如数据库读取。我想开始一个新 Activity 并在所有异步任务完成后呈现其内容。实现这一目标的最佳方法是什么？ 我知道 onPostExecute
我有一个脚本需要命令名称和该命令的参数作为参数。 所以我想编写一个完成函数来完成命令的名称并完成该命令的参数。 所以我可以这样完成命令的名称 if [[ "$COMP_CWORD" == 1 ]];
我的应用程序有一个相当奇怪的行为。我在 BOOT_COMPLETE 之后启动我的应用程序，因此在我启动设备后它是可见的。 GUI 响应迅速，一切正常，直到我调用 finish()，按下按钮时，什么都没
我是一名优秀的程序员,十分优秀！