Mule DB 数据检索成块

Question

我们正在尝试提取大约。来自数据库的 40 GB 数据并希望生成多个 csv 文件。我们以流方式使用 mule DB 连接器，它返回 'ResultSetIterator'

Q1) 如何将此ResultSetIterator 转换为arraylist？或任何我们可以进一步用于生成文件的可读格式

Q2）我们尝试使用 For-Each 组件将这些数据拆分成块，它适用于有限的数据集和大量数据，提供SerializationException

在下面的输入 sn-p 中，我们使用 for-each 制作数据块并将其提供给多个文件的批处理

  <batch:job name="testBatchWithDBOutside">
        <batch:input>
            <logger message="#[payload]" level="INFO" doc:name="Logger"/>
        </batch:input>
        <batch:process-records>
            <batch:step name="Batch_Step">
                <batch:commit size="10" doc:name="Batch Commit">
                    <object-to-string-transformer doc:name="Object to String"/>
                    <logger message="#[payload]" level="INFO" doc:name="Logger"/>
                    <file:outbound-endpoint path="C:\output" outputPattern="#[message.id].txt" responseTimeout="10000" doc:name="File"/>
                </batch:commit>
            </batch:step>
        </batch:process-records>
    </batch:job>
    <flow name="testBatchWithDBOutsideFlow" processingStrategy="synchronous">
        <file:inbound-endpoint path="C:\input" responseTimeout="10000" doc:name="File"/>
        <db:select config-ref="MySQL_Configuration" streaming="true" fetchSize="10" doc:name="Database">
            <db:parameterized-query><![CDATA[select * from classicmodels]]></db:parameterized-query>
        </db:select>
        <foreach batchSize="5" doc:name="For Each">
            <batch:execute name="testBatchWithDBOutside" doc:name="testBatchWithDBOutside"/>
        </foreach>
    </flow>

Answer 1

第一季度。您不想将 Iterator 转换为 List，因为这会破坏从 DB 连接器进行流式传输并将所有记录加载到内存中的目的。无论如何，Mule 以相同的方式处理迭代器和列表。

第二季度。批处理模块意味着一个 for-each 操作。 batch:input 的输出需要是 List 或 Iterator。你应该能够简化这个

<batch:job name="testBatch">
    <batch:input>
        <db:select config-ref="MySQL_Configuration" streaming="true" fetchSize="10" doc:name="Database">
            <db:parameterized-query><![CDATA[select * from classicmodels]]></db:parameterized-query>
        </db:select>
    </batch:input>
    <batch:process-records>
        <batch:step name="Batch_Step">
            <object-to-string-transformer doc:name="Object to String"/>
            <file:outbound-endpoint path="C:\output" outputPattern="#[message.id].txt" responseTimeout="10000" doc:name="File"/>
        </batch:step>
    </batch:process-records>
</batch:job>

您还需要将 object-to-string-transformer 替换为转换数据库记录的组件（此时的有效负载将是一个映射，其中键是列名，值是记录值) 转换成 csv 行。

您可以在 Mule 博客中找到一个不错的示例：https://blogs.mulesoft.com/dev/anypoint-platform-dev/batch-module-reloaded/

另一种选择是移除批处理器并使用 DataWeave 生成​​ csv 输出并将其流式传输到文件中。这可能会有所帮助：https://docs.mulesoft.com/mule-user-guide/v/3.7/dataweave-streaming

Dataweave 将在处理每条记录时调用 ResultSetIterator 上的 next，并且该 Iterator 将处理从底层数据库中选择记录块，因此步骤之间没有排队，也没有将完整的数据集加载到内存中。

<flow name="batchtestFlow">
    <http:listener config-ref="HTTP_Listener_Configuration" path="/batch" allowedMethods="GET" doc:name="HTTP"/>
    <db:select config-ref="Generic_Database_Configuration" streaming="true" doc:name="Database">
        <db:parameterized-query><![CDATA[select * from Employees]]></db:parameterized-query>
    </db:select>
    <dw:transform-message doc:name="Transform Message">
        <dw:set-payload><![CDATA[%dw 1.0
            %input payload application/java
            %output application/csv streaming=true, header=true, quoteValues=true
            ---
            payload map ((e, i) -> {
                surname: e.SURNAME,
                firstname: e.FIRST_NAME
            })]]></dw:set-payload>
    </dw:transform-message>
    <file:outbound-endpoint path="C:/tmp" outputPattern="testbatchfile.csv" connector-ref="File" responseTimeout="10000" doc:name="File"/>
</flow>

Answer 2

您想使用 OutputHander。确保您已打开流式传输，然后使用脚本组件，例如选择 groovy 并一次处理每一行，如下所示：

// script.groovy
return {evt, out ->

 payload.each { row ->
  out << row.SOMECOLUMN....  }

} as OutputHandler

还有你 xml 中的组件

    <scripting:transformer returnClass="TODO" doc:name="ScriptComponent">
        <scripting:script engine="Groovy" file="script.groovy" />
    </scripting:transformer>

如果你想返回一些输出。但是，如果您想在您的情况下写入文件，则不会使用变量 out，而是写入您的文件。

Answer 3

我找到了一个简单快捷的方法如下：

这里 DB 连接器处于流模式，并且 For-Each 以给定的 Batch Size 拆分记录

<flow name="testFlow" processingStrategy="synchronous">
        <composite-source doc:name="Composite Source">
            <quartz:inbound-endpoint jobName="test" cronExpression="0 48 13 1/1 * ? *" repeatInterval="0" connector-ref="Quartz" responseTimeout="10000" doc:name="Quartz">
                <quartz:event-generator-job/>
            </quartz:inbound-endpoint>
            <http:listener config-ref="HTTP_Listener_Configuration" path="/hit" doc:name="HTTP"/>
        </composite-source>
        <db:select config-ref="MySQL_Configuration" streaming="true" fetchSize="10000" doc:name="Database">
            <db:parameterized-query><![CDATA[SELECT * FROM tblName]]></db:parameterized-query>
        </db:select>
         <foreach batchSize="10000" doc:name="For Each">

                    <dw:transform-message doc:name="Transform Message">
                                           <dw:set-payload><![CDATA[%dw 1.0
                   %output application/csv
                   ---
                   payload map {
                    field1:$.InterfaceId,
                    field2:$.Component

                   }]]></dw:set-payload>
                    </dw:transform-message>
               <file:outbound-endpoint path="F:\output" outputPattern="#[message.id].csv" responseTimeout="10000" doc:name="File"/>
        </foreach>
        <set-payload value="*** Success ***" doc:name="Set Payload"/>

    </flow>