Load a CSV file

id,amount,currency,trader,traderEntity,bank
1,12394,chf,E F,E3,BNP Paribas
2,13149,gbp,A B,E4,BNP Paribas

final ICSVSource<Path> csvSource = CsvSourceFactory.create();

final CSVParserConfiguration parserConfig = new CSVParserConfiguration(columnNames);
// Default separator is ',', you can customize this.
parserConfig.setSeparator(',');
// Skip the first header line if your file has a header.
parserConfig.setNumberSkippedLines(1);
// You can override the default charset
parserConfig.setCharset(StandardCharsets.UTF_8);
// Accept ignored supplementary columns
parserConfig.setAcceptOverflowingLines(true);

final String topicName = storeName;
// Will be close by source
final SingleFileCSVTopic csvTopic =
    new SingleFileCSVTopic(topicName, parserConfig, filePath, null);
csvSource.addTopic(csvTopic);

final IStoreMessageChannelFactory<IFileInfo<Path>, ILineReader> channelFactory =
    new CSVMessageChannelFactory<>(csvSource, datastore);

final Fetch<IFileInfo<Path>, ILineReader> fetch = new Fetch<>(channelFactory);
fetch.fetch(csvSource);

final IStoreMessageChannel<IFileInfo<Path>, ILineReader> channel =
    channelFactory.createChannel(STORE_NAME, STORE_NAME);
datastore.getTransactionManager().startTransaction();
csvSource.fetch(List.of(channel));
datastore.getTransactionManager().commitTransaction();

csvSource.close();

final AColumnCalculator<ILineReader> calculator =
    new AColumnCalculator<>(negatedAmountField) {
      @Override
      public Object compute(final IColumnCalculationContext<ILineReader> context) {
        final Long amountValue = (Long) context.getValue(AMOUNT_FIELD);
        return Math.negateExact(amountValue);
      }
    };

channelFactory.setCalculatedColumns(STORE_NAME, List.of(calculator));

final Fetch<IFileInfo<Path>, ILineReader> fetch = new Fetch<>(channelFactory);
fetch.fetch(csvSource);