Mule 批量提交并记录失败

Question

我目前的情况:

我有 10000 条记录作为批处理的输入 .
根据我的理解，批处理仅用于逐条记录处理。因此，我在批处理步骤中使用 dataweave 组件转换每条记录(注意:我没有使用任何批处理提交)并将每条记录写入文件。进行逐条记录处理的原因是假设在任何特定记录中，都有一个 无效 数据，只有该特定记录失败，其余的将被很好地处理。

但是在我看到的许多博客中，他们使用的是 批量提交(带流媒体)带有数据编织组件。所以根据我的理解，所有的记录都会一次性给到dataweave，如果一个记录有无效数据，所有10000条记录都会失败(在dataweave)。这样，逐条记录处理的点就丢失了。
以上假设是正确的还是我想错了？

这就是我不使用批处理提交的原因。

现在，正如我所说，将每条记录发送到一个文件中。实际上，我确实需要将每条记录发送到 5 个不同的 CSV 文件 .所以，目前我正在使用 分散-聚集 我的 中的组件BatchStep 将其发送到五个不同的路线。

因为，你可以看到图像。输入阶段给出了 10000 条记录的集合。每个记录将使用 Scatter-Gather 发送到 5 个路由。

是，我使用的方法可以吗，或者可以遵循更好的设计？

此外，我创建了第二批步骤，以捕获 只有失败的记录。 但是，使用当前的设计，我无法捕获失败的记录。

Answer 1

简短的答案

Is the above assumption correct or am I thinking wrong way??

简而言之，是的，您的想法是错误的。阅读我的 loooong 示例说明以了解原因，希望您会喜欢。

Also, I have created a 2nd Batch step, to capture ONLY FAILEDRECORDS. But, with the current Design, I am not able to Capture failed records.

你可能忘记设置 max-failed-records = "-1" (无限制)批处理作业。默认为 0，第一个失败的记录批处理将返回并且不执行后续步骤。

Is, the approach I am using is it fine, or any better Design can be followed??

我认为如果性能对您至关重要，并且您无法应对按顺序执行此操作所产生的开销，这是有道理的。
相反，如果您可以放慢一点，那么在 5 个不同的步骤中执行此操作可能是有意义的，您将失去并行性，但您可以更好地控制失败的记录，尤其是在使用批量提交时。

MULE 批量作业在实践中

我认为最好的方法是通过一个例子来解释它是如何工作的。

考虑以下情况:
你有一个批处理配置 max-failed-records = "-1" (没有限制)。

<batch:job name="batch_testBatch" max-failed-records="-1">

在这个过程中我们输入了一个由 6 个字符串组成的集合。

 <batch:input>
            <set-payload value="#[['record1','record2','record3','record4','record5','record6']]" doc:name="Set Payload"/>
 </batch:input>

处理由3个步骤组成"
第一步只是记录处理过程，第二步将改为进行记录并在 record3 上抛出异常以模拟失败。

<batch:step name="Batch_Step">
        <logger message="-- processing #[payload] in step 1 --" level="INFO" doc:name="Logger"/>
 </batch:step>
 <batch:step name="Batch_Step2">
     <logger message="-- processing #[payload] in step 2 --" level="INFO" doc:name="Logger"/>
     <scripting:transformer doc:name="Groovy">
         <scripting:script engine="Groovy"><![CDATA[
         if(payload=="record3"){
             throw new java.lang.Exception();
         }
         payload;
         ]]>
         </scripting:script>
     </scripting:transformer>
</batch:step>

第三步将只包含提交计数值为 2 的提交。

<batch:step name="Batch_Step3">
    <batch:commit size="2" doc:name="Batch Commit">
        <logger message="-- committing #[payload] --" level="INFO" doc:name="Logger"/>
    </batch:commit>
</batch:step>

现在你可以跟着我执行这个批处理:



开始时，第一步将处理所有 6 条记录，登录控制台将如下所示:

 -- processing record1 in step 1 --
 -- processing record2 in step 1 --
 -- processing record3 in step 1 --
 -- processing record4 in step 1 --
 -- processing record5 in step 1 --
 -- processing record6 in step 1 --
Step Batch_Step finished processing all records for instance d8660590-ca74-11e5-ab57-6cd020524153 of job batch_testBatch

现在事情在第 2 步中会更有趣，记录 3 将失败，因为我们明确抛出异常，但尽管如此，该步骤仍将继续处理其他记录，这里的日志将是什么样子。

-- processing record1 in step 2 --
-- processing record2 in step 2 --
-- processing record3 in step 2 --
com.mulesoft.module.batch.DefaultBatchStep: Found exception processing record on step ...
Stacktrace
....
-- processing record4 in step 2 --
-- processing record5 in step 2 --
-- processing record6 in step 2 --
Step Batch_Step2 finished processing all records for instance d8660590-ca74-11e5-ab57-6cd020524153 of job batch_testBatch

此时尽管在这一步中记录失败，批处理仍将继续，因为参数 max-failed-records设置为 -1 (无限制)而不是默认值 0。

此时 所有成功记录将被传递到第 3 步，这是因为，默认情况下， accept-policy步骤的参数设置为 NO_FAILURES . (其他可能的值是 ALL 和 ONLY_FAILURES )。

现在包含计数等于 2 的提交阶段的 step3 将两两提交记录:

-- committing [record1, record2] --
-- committing [record4, record5] --
Step: Step Batch_Step3 finished processing all records for instance d8660590-ca74-11e5-ab57-6cd020524153 of job batch_testBatch
-- committing [record6] --

如您所见，这证实了失败的 record3 没有传递到下一步，因此没有提交。

从这个例子开始，我认为你可以想象和测试更复杂的场景，例如在提交之后你可以有另一个步骤来处理失败的记录，让管理员知道失败的邮件。
在您始终可以使用外部存储来存储有关您的记录的更多高级信息之后，您可以在 my answer to this other question 中阅读.

希望这可以帮助