How to load CSV files

A CSV topic reads delimited text files and feeds their rows to channels. The CSV family is configured through ICsvTopicConfigurator, which is shared by all CSV topics, local and cloud. ILocalCsvTopicConfigurator reads from a local directory. For the general topic and channel concepts, see the data loading overview.

How to load CSV files from a directory

Call IDataLoadConfigurator.channels().addChannel(channelName, tableName) to declare the channel that loads rows into the target table.
Call IDataLoadConfigurator.topics().addTopic() with ILocalCsvTopicConfigurator.class to declare a local CSV topic.
Call path() to point the topic at a directory, and filePattern() to select files. The pattern accepts a glob: or a regex: prefix. Use a recursive pattern such as glob:**/*.csv to match files in subdirectories.
Call attachChannel(channelName) on the topic to feed the channel.

The following example loads every CSV file in /data/sales into the Sales table:

@Bean
public IDataLoadConfigurer loadCsvDirectoryConfigurer() {
  return configurator -> {
    configurator.channels().addChannel("SalesChannel", "Sales");
    configurator
        .topics()
        .addTopic("sales-csv", ILocalCsvTopicConfigurator.class)
        .path("/data/sales")
        .filePattern("glob:*.csv")
        // Define a parser for every column whose name matches a field of the target table.
        .parsersFromTable("Sales")
        .attachChannel("SalesChannel");
  };
}

How are headers mapped to table fields?

The loader reads each file’s header row first, then routes column values to the table fields by name. As a result:

Files inside the same topic can list their columns in a different order.
Extra columns in a file are ignored.
A file can contain only a subset of the expected columns.

// The Sales table has three fields: productId, productName, amount.
// Headers are matched to those fields by name; order in the file
// doesn't matter, missing columns are loaded as null, and any
// column not declared on the table is silently dropped.
//
// File A header: productId,productName,amount             -> all three fields populated
// File B header: amount,productId,productName             -> same; order does not matter
// File C header: productId,productName                    -> amount loaded as null
// File D header: productId,productName,amount,internalRef -> internalRef dropped
configurator.channels().addChannel("SalesChannel", "Sales");
configurator
    .topics()
    .addTopic("sales-csv", ILocalCsvTopicConfigurator.class)
    .path("/data/sales")
    .filePattern("glob:*.csv")
    .parsersFromTable("Sales")
    .attachChannel("SalesChannel");

How to configure CSV column parsers

CSV topics do not automatically infer column parsers from the target table’s schema. Define them explicitly. Columns left without a parser are parsed as String. The simplest approach is to define parsers from a database table in one call. Every column whose name matches a field of the table is parsed with that field’s type:

parsersFromTable(tableName): define parsers for every column that matches a field of the given table. This overwrites any parser previously configured for those columns.
columnParser(column, parserKey): set the parser for a single column, using a registered IParser plugin key.
columnParser(column, Function): set an ad-hoc parser for a single column without registering a plugin. The function turns the raw CSV cell into the parsed value.

Order matters. Each of these methods overwrites any parser previously set for its column, so the last call wins. To make a per-column override win, call columnParser after parsersFromTable. Parsers are applied after column renaming. A parser column key must therefore be the post-rename name. Configure addColumnRenames before parsersFromTable.

configurator
    .topics()
    .topic("sales-csv", ICsvTopicConfigurator.class)
    .parsersFromTable("Sales")
    // cobDate is not a field of the Sales table, so parsersFromTable does not define it;
    // define an ad-hoc parser for it explicitly.
    .columnParser("cobDate", seq -> LocalDate.parse(seq, DateTimeFormatter.BASIC_ISO_DATE));

How to tune CSV parsing

ILocalCsvTopicConfigurator exposes the usual CSV parser options:

separator(char): field delimiter.
charset(Charset): file encoding.
processQuotes(boolean): Excel-style quote handling.
fileColumnNames(List<String>): provide column names for files that have no header row.

configurator
    .topics()
    .topic("sales-csv", ICsvTopicConfigurator.class)
    .path("/data/sales")
    .filePattern("glob:*.csv")
    .separator(';')
    .charset(StandardCharsets.ISO_8859_1)
    .processQuotes(true);

CSV global settings

ICsvTopicConfigurator declares CSV-specific SettingKey constants that apply to every CSV topic, local and cloud:

PARSER_THREADS
DEFAULT_CHARSET
DEFAULT_SEPARATOR
ENABLE_PARSING_REPORT
BUFFER_SIZE

The following example raises the number of CSV parser threads from the default of 4 to 8:

configurator.globalSettings().set(ICsvTopicConfigurator.PARSER_THREADS, 8);

Load data: general topic, channel, and field calculator concepts.