如何仅在另一个处理器未执行时执行处理器?

问题描述

我正在向表中插入/更新数据.数据库系统不提供Upsert"功能.因此，我使用临时表进行插入，然后合并到最终"表中，最后我截断了临时表.

这会导致竞争条件.如果在合并+截断之间将新数据插入临时表，则此数据将丢失.

我如何确保不会发生这种情况?

我尝试通过等待/通知对此进行建模，但这也不是一个干净的解决方案.Put Data into staging table"PutDatabaseRecord 处理器的队列可以被填满，MergeVertica for Insert/Update"ExecuteSQL 仍然可以执行.

解决方案

我会使用具有 60 或 30 秒阈值的 MonitorActivity 处理器，并使用带有 Continually Send Messages 的 Inactive 输出 设置为假".

成功将 SQL 插入暂存连接到您的 MonitorActivity，这样如果在最后 X 秒内没有看到任何活动，他将触发一个流文件，该流文件将启动您的合并过程.

从

I am inserting/updating data into a table. The database system does not provide an "Upsert" functionality. Thus I am using a staging table for the insert followed by a merge into the "final" table and finally I am truncating the staging table.

This leads to a race condition. If new data is inserted into the staging table between the merge+truncate this data is lost.

How can I make sure this does not happen?

I have tried to model this via Wait/Notify, but this is not a clean solution either. The queue for the "Put Data into staging table" PutDatabaseRecord processor could be filled and "MergeVertica for Insert/Update" ExecuteSQL could still execute.

解决方案

I would use a MonitorActivity processor with a 60 or 30 sec threshold and use the Inactive output with a Continually Send Messages set to "false".

Have the success of the SQL inserts into staging connection into your MonitorActivity, this way if no activity is seen in the last X seconds he will trigger a flowfile that will start your Merge Process.

Download the template from https://codeshare.io/aJNNkn

这篇关于如何仅在另一个处理器未执行时执行处理器?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！