Camel - стрим: загрузка файла и ftp

Question

У меня есть приложение Camel / Spring Boot, которое извлекает данные из конечной точки GraphQL, сохраняет данные в базе данных в памяти (2 таблицы), извлекает файл CSV, выполняя запрос SQL, а затем должен загрузить файлна FTP-сервер.Поскольку есть ~ 350 тыс. Записей, которые будут извлечены, я использую SQL 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.camel.InvalidPayloadException: тело недоступно типа: java.io.InputStream, но имеет значение: org.apache.camel.component.sql.ResultSetIterator Я полагаю, это связано с потоковой передачей файлов.Есть ли способ выполнить загрузку по ftp после того, как поток: файл завершен и у него есть последующий маршрут на «closeOnDone»?

В другой заметке мне было бы интересно, как 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}");