How to load data - Atoti Hub

Load data into the database using the IDataLoadConfigurer interface. A data load configuration declares three kinds of objects:

Channels define how rows land in a target table.
Topics define where the data comes from, for example a directory of CSV files.
Field calculators are optional and derive new fields from the source columns.

An optional data-operation trigger can also run inside the load transaction to derive extra rows, for example to maintain a data-availability table atomically with the load.

Prerequisites

Loading data requires:

A configured database.
A configured cube.

What is the IDataLoadConfigurer interface?

Data loading is configured through a Spring bean that implements the IDataLoadConfigurer interface. The configure() method receives an IDataLoadConfigurator. Use it to declare topics, channels, and field calculators. See the Javadoc for the complete list of available methods.

@Bean
public IDataLoadConfigurer dataLoadConfigurer() {
  return _ -> {
    // Declare topics, channels and calculators on the configurator.
  };
}

What are topics and channels?

A channel defines how rows load into a specific target table. It carries the field calculators, column renames, tuple publisher, and unload condition for that table. Channels are created on the configurator, and each one has a unique name. A topic describes a data source, for example a directory of CSV files together with the parser options needed to read them. A topic feeds one or more channels through attachChannel(channelName). The same channel can be attached to several topics. Within a single topic, two attached channels cannot target the same table. The following example puts both objects together. The topic reads CSV files from a directory and feeds their rows to the SalesChannel, which loads them 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");
  };
}

This example uses a CSV topic. For the full CSV reference, covering file selection, parsers, and parsing options, see the CSV topic page.

How to rename columns

Use addColumnRenames() to map header names onto the expected field names. Renames declared on the topic apply to every channel. Renames declared on a channel apply on top of them.

configurator
    .topics()
    .topic("sales-csv", ICsvTopicConfigurator.class)
    .addColumnRenames(Map.of("_raw_price_", "amount"));

How to compute a derived field with a field calculator

A field calculator produces a field whose value is computed from other source columns. To create one, implement IFieldCalculator:

getFieldName(): name of the computed field.
getType(): type of the computed value.
getUnderlyingFieldNames(): names of the source columns the calculator depends on.
compute(): writes the computed value.
onMissingFields(): optional. Decides what to do when an underlying column is missing. The default throws.

A calculator can be registered at two levels:

On the table, through configurator.addCalculator(tableName, calculator). Every channel that targets that table inherits the calculator.
On a single channel, by chaining off configurator.channels().addChannel(...) at creation time, or through configurator.channels().channel(channelName).addCalculator(calculator) afterward.

When a channel-level calculator and a table-level calculator share the same field name, the channel-level one takes precedence.

@Bean
public IDataLoadConfigurer fieldCalculatorConfigurer() {
  return configurator -> {
    configurator.addCalculator("Sales", new ProductNameCalculator());
    configurator.channels().addChannel("SalesChannel", "Sales");
    configurator
        .topics()
        .addTopic("sales-csv", ILocalCsvTopicConfigurator.class)
        .path("/data/sales")
        .filePattern("glob:*.csv")
        .parsersFromTable("Sales")
        .attachChannel("SalesChannel");
  };
}
/** Derives {@code productName} from the {@code productId} column. */
public static class ProductNameCalculator implements IFieldCalculator {
  @Override
  public String getFieldName() {
    return "productName";
  }
  @Override
  public ContentType getType() {
    return ContentType.STRING;
  }
  @Override
  public List<String> getUnderlyingFieldNames() {
    return List.of("productId");
  }
  @Override
  public void compute(
      final IArrayReader underlyingFieldReader,
      final IFieldContext context,
      final IWritableCell output) {
    output.write("name " + underlyingFieldReader.read(0));
  }
}

How to handle missing source fields

Override onMissingFields() on the calculator to either tolerate missing source columns or fail fast when a column is required. Tolerating missing columns helps when files in the same topic carry different subsets of columns.

/** Writes {@code "UNKNOWN"} when the underlying {@code colorCode} is absent from the file. */
public static class ProductColorCalculator implements IFieldCalculator {
  @Override
  public String getFieldName() {
    return "productColor";
  }
  @Override
  public ContentType getType() {
    return ContentType.STRING;
  }
  @Override
  public List<String> getUnderlyingFieldNames() {
    return List.of("colorCode");
  }
  @Override
  public void onMissingFields(final Collection<String> missingFields) {
    // Override the default (which throws) to accept files that do not
    // carry the colorCode column.
  }
  @Override
  public void compute(
      final IArrayReader underlyingFieldReader,
      final IFieldContext context,
      final IWritableCell output) {
    if (underlyingFieldReader.isNull(0)) {
      output.write("UNKNOWN");
    } else {
      output.write(underlyingFieldReader.read(0));
    }
  }
}

How to load one topic into several tables

Declare one channel per target table on the configurator, then call attachChannel() on the topic for each one. The topic fans its rows out to every attached channel.

configurator.channels().addChannel("SalesChannel", "Sales");
configurator.channels().addChannel("SalesAuditChannel", "SalesAudit");
configurator
    .topics()
    .addTopic("sales-csv", ILocalCsvTopicConfigurator.class)
    .path("/data/sales")
    .filePattern("glob:*.csv")
    // Define parsers from both target tables. If the two tables share a column with different
    // parsers, the second call wins (last write).
    .parsersFromTable("Sales")
    .parsersFromTable("SalesAudit")
    .attachChannel("SalesChannel")
    .attachChannel("SalesAuditChannel");

How to override topic settings

IDataLoadConfigurator.globalSettings() overrides the defaults used by topics. Each setting is declared as a SettingKey constant on the topic configurator interface it applies to. A setting only affects topics whose configurator implements that interface. For example, ICsvTopicConfigurator declares PARSER_THREADS, DEFAULT_CHARSET, and DEFAULT_SEPARATOR. These apply to every CSV topic, such as local and cloud, and are ignored by topics of other types. 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 CSV files: file selection, column parsers, and parsing options for the CSV topic.
Load rows from a SQL query: JDBC connections, column aliasing, and query parameters.
Load tuples from a supplier: rows built in code, including tuples derived from a datastore query.
Data operation triggers: run custom logic inside the load transaction, for example to maintain a data-availability table.

​Prerequisites

​What is the IDataLoadConfigurer interface?

​What are topics and channels?

​How to rename columns

​How to compute a derived field with a field calculator

​How to handle missing source fields

​How to load one topic into several tables

​How to override topic settings

​Related pages