骆驼 - 流：文件和 FTP 上传

Question

我有一个Camel/SpringBoot应用程序，它从GraphQLendpoint检索数据，将数据存储在内存数据库（2个表）中，通过运行SQL查询提取CSV文件，然后将文件上传到FTP服务器。由于将提取约350k条记录，我使用SQLs outputType=StreamList、splitter和stream:file。整个路线如下所示：

from("direct:loadComplete").id("loadComplete")
    .log("${date:now:yyyy-MM-dd'T'HH:mm:ssZ} - Load complete")
    .setHeader("createDate", simple("${date:now:yyyyMMdd'_'HHmmss}"))
    .setHeader(Exchange.FILE_NAME,simple("${header.createDate}_sku_{{account.countryCode}}.csv"))
    .to("sql:"+ QUERY + "?outputType=StreamList") //run the SQL query
    .split(body()).streaming() //stream & split the results
        .bean(LineProcessor.class, "processLine") //process each entry and transform to CSV
        .toD("stream:file?fileName={{setting.outputDir}}/${header.CamelFileName}&closeOnDone=true") // stream each record to file
    .end()
    .log("${date:now:yyyy-MM-dd'T'HH:mm:ssZ} - Written file to disk:${header.CamelFileName}")
    .to("ftp://{{ftp.host}}:{{ftp.port}}/{{ftp.path}}?fileName={{setting.outputDir}}/${header.CamelFileName}&username={{ftp.username}}&password={{ftp.password}}&{{ftp.config}}") // upload the CSV file
    .log("${date:now:yyyy-MM-dd'T'HH:mm:ssZ} - FTP upload complete. File: ${header.CamelFileName}");

提取数据时不会出现任何问题，并使用记录创建CSV文件。但是，上传到FTP服务器失败，原因是：org.apache.camale。InvalidPayloadException:没有类型为java.io的可用正文。InputStream，但具有值：org.apache.camel.component.sql。ResultSetIterator我相信这是由于文件流。一旦流：文件完成并在“closeOnDone”上有后续路径，是否有任何方法执行ftp上传？

另一方面，我对outputType=StreamList在后台的实际工作方式很感兴趣。我假设它仍然将完整的SQL结果加载到内存中？有没有其他方法可以最大限度地减少应用程序的内存消耗，并且不需要传递大量的有效负载？有没有其他更优雅的方法来使用CSV编组？

谢谢你的帮助！

霍尔格

Answer 1

在流处理中，将创建的 csv 行上传到本地文件。在 “.end（）” 之后，正文将包含 sql 生产者的结果。您应该从FTP生产者的本地文件中读取数据。

from("direct:loadComplete").id("loadComplete")
            .log("${date:now:yyyy-MM-dd'T'HH:mm:ssZ} - Load complete")
            .setHeader("createDate", simple("${date:now:yyyyMMdd'_'HHmmss}"))
            .setHeader(Exchange.FILE_NAME,simple("${header.createDate}_sku_{{account.countryCode}}.csv"))
            .to("sql:"+ QUERY + "?outputType=StreamList") //run the SQL query
            .split(body()).streaming() //stream & split the results
            .bean(LineProcessor.class, "processLine") //process each entry and transform to CSV
            .toD("stream:file?fileName={{setting.outputDir}}/${header.CamelFileName}&closeOnDone=true") // stream each record to file
            .end()
            .log("${date:now:yyyy-MM-dd'T'HH:mm:ssZ} - Written file to disk:${header.CamelFileName}")
            .pollEnrich("file:{{setting.outputDir}}?fileName=${header.CamelFileName}&noop=true", 10000/*investigate about timing according your file size*/, new AggregationStrategy() {
                @Override
                public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
                    return newExchange;// save important data from old exchange
                }
            })
            .to("ftp://{{ftp.host}}:{{ftp.port}}/{{ftp.path}}?fileName={{setting.outputDir}}/${header.CamelFileName}&username={{ftp.username}}&password={{ftp.password}}&{{ftp.config}}") // upload the CSV file
            .log("${date:now:yyyy-MM-dd'T'HH:mm:ssZ} - FTP upload complete. File: ${header.CamelFileName}");