Пропустите первую строку csv, используя верблюжий сплиттер EIP, обработайте все остальные строки и объедините все строки, включая пропущенную строку
Есть ли простой способ пропустить первую строку csv, header, используя верблюжий сплиттер EIP, обработать все остальные строки и объединить все строки, включая пропущенную строку? Мне нужно преобразовать дату каждой записи в файл CSV, но пропустить первую строку, которая является заголовком. Я пытаюсь использовать верблюжий EIP сплиттер. благодарю вас!
<route
id="core.predix.consumer.route"
autoStartup="true" >
<from id="predixConsumer" ref="predixConsumer" />
<convertBodyTo type="java.lang.String" />
<split streaming="true" > <!-- strategyRef="enrichmentAggregationStrategy" stopOnException="true"> -->
<tokenize token="\n"/>
<log message="Split line ${body}"/>
<!-- <process ref="EnrichementProcessor"/> -->
</split>
<to uri="{{fileDestinationEndpoint}}" />
</route>
3 ответа
Немного поздно, но...
поставить логику в агрегаторе...
<route
id="fadec.core.PostFlightReportProducerRoute"
autoStartup="true">
<from uri="seda:postFlightReportProducer" />
<split streaming="true" strategyRef="statusAggregationStrategy">
<simple>${body}</simple>
<log message="inbound Post Flight Summary Report message body: ${body}" />
<process ref="postFlightReportMarshaler" />
<to uri="velocity:templates/postFlightReportSummary.vm" />
<log message="Velocity output: ${body}" loggingLevel="INFO" />
</split>
<wireTap uri="{{DFEndpointTest}}fileName=/${header.aircraftMetadata.outputPath}/${header.ship}_${date:now:yyyy-MM-dd_HH-mm-ss}.csv" />
<choice>
<when>
<simple>${body} == null</simple>
<log message="body is NULL, do not send NULL body!" />
<stop></stop>
</when>
<otherwise>
<process ref="xlsxProcessor"/>
<wireTap uri="{{DFEndpointTest}}fileName=/${header.aircraftMetadata.outputPath}/${header.ship}_${date:now:yyyy-MM-dd_HH-mm-ss}.xlsx" />
<log message="Sending data packet: ${header.aircraftMetadata.outputPath}/${header.ship}_${date:now:yyyy-MM-dd_HH-mm-ss}.xlsx" />
<stop></stop>
</otherwise>
</choice>
</route>
public class StatusAggregationStrategy implements AggregationStrategy {
private Logger log = LoggerFactory.getLogger(StatusAggregationStrategy.class.getName());
@Override
public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
//-------------------------------------------------------------------------------------
// Arrived | oldExchange | newExchange | Description
//-------------------------------------------------------------------------------------
// A | NULL | A | first message arrives for the first group
// B | A | B | second message arrives for the first group
// F | NULL | F | first message arrives for the second group
// C | AB | C | third message arrives for the first group
//---------------------------------------------------------------------------------------
log.info("Status Aggregation Strategy :: start");
if ( oldExchange == null ) { //This will set the 1st record with the Header
log.info("old Exchange is Null");
String body = newExchange.getIn().getBody(String.class);
newExchange.getIn().setBody(body);
return newExchange;
}
//Each newBody msg exchange will have 2 records, a header and a data record
String newBody = newExchange.getIn().getBody(String.class);
String existingBody = oldExchange.getIn().getBody(String.class);
StringBuilder osb = new StringBuilder();
log.info("New Body exchange: " + newBody);
log.info("Old Body exchange: " + existingBody);
String SkipRecord = "";
String addRecord = "";
Scanner osc = new Scanner(newBody).useDelimiter("\r\n|\n");
while ( osc.hasNextLine() ) {
SkipRecord = osc.nextLine();
//osc.nextLine(); //move past header
log.info("aggregation: skip record: " + SkipRecord);
if ( osc.hasNextLine() ) {
addRecord = osc.nextLine();
log.info("aggregation addRecord: " + addRecord );
osb.append(addRecord).append(System.lineSeparator());
} else { log.error("bad newBody message exchange, has no data record!"); }
}
osc.close();
log.info("Joined exchange: Old body: " + existingBody + " New body: " + osb.toString());
oldExchange.getIn().setBody(existingBody + osb.toString() );
log.debug("Status Aggregation Strategy :: finish");
return oldExchange;
} //Exchange process
Это не работает Я думаю, что это будет работать, потому что он должен выполнять только то, что находится внутри оператора, если property.CamelSlitIndex > 0, но он выполняется для каждой строки, даже когда CamelSplitIndex =- 0.
это ошибка, или я использую ее неправильно???
<route
id="core.predix.consumer.route"
autoStartup="true" >
<from id="predixConsumer" ref="predixConsumer" />
<convertBodyTo type="java.lang.String" />
<split streaming="true" > <!-- strategyRef="enrichmentAggregationStrategy" stopOnException="true"> -->
<tokenize token="\n"/>
<log message="Split line ${body}"/>
<choice>
<when>
<simple>"${property.CamelSplitIndex} > 0"</simple>
<process ref="timeStampEnrichmentProcessor" />
<log message="Camel Split Index is greater than zero: ${property.CamelSplitIndex}" />
</when>
</choice>
</split>
<log message="body: ${body}" />
<to uri="{{fileDestinationEndpoint}}" />
</route>
Вы можете использовать опцию org.apache.camel.dataformat.csv.CsvDataFormat#setSkipHeaderRecord(true)
Вот как это может работать:
package app;
import io.github.dgroup.j2gp.processor.GroupByParent;
import org.apache.camel.CamelContext;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.dataformat.csv.CsvDataFormat;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.commons.csv.CSVFormat;
import java.util.List;
public class App {
public static void main(String[] args) throws Exception {
CamelContext ctx = new DefaultCamelContext();
ctx.addRoutes(new RouteBuilder() {
public void configure() {
CsvDataFormat csv = new CsvDataFormat();
csv.setFormat(CSVFormat.EXCEL);
csv.setSkipHeaderRecord(true); // <- ignore first line (e.g. csv header)
from("file:data/inbox?noop=true")
.unmarshal(csv)
.split(body())
.log("Reached ${body}")
.process(exchange -> {
List<String> line = exchange.getIn().getBody(List.class);
...
});
...
}
});
// start the route and let it do its work
ctx.start();
Thread.sleep(10000);
// stop the CamelContext
ctx.stop();
}
}