Batch Messages

Warning

The BATCH message functionality is currently in preview and is subject to change. You can open an issue or join the discussion on GitHub to provide feedback during the preview period.

The Singer message specification defines the three basic types of messages: RECORD, STATE, and SCHEMA. The RECORD message is used to send data from the tap to the target. The STATE message is used to send state data from the tap to the target. The SCHEMA message is used to send schema data from the tap to the target, and for example, create tables with the correct column types.

However, the Singer specification can be extended to support additional types of messages. For example, the ACTIVATE_VERSION message is used to manage hard deletes in the target.

This library’s implementation of the BATCH message is used to send records in bulk from the tap to the target, using an intermediate filesystem to store batch files. This is useful, for example

  • when the tap outputs records at a much higher rate than the target can consume them, creating backpressure

  • when the source system can directly export data in bulk (e.g. a database dump)

Currently only a local filesystem is supported, but other filesystems like AWS S3, FTP, etc. could be supported in the future.

The BATCH Message

{
  "type": "BATCH",
  "stream": "users",
  "encoding": {
    "format": "jsonl",
    "compression": "gzip"
  },
  "manifest": [
    "file://path/to/batch/file/1",
    "file://path/to/batch/file/2"
  ]
}

encoding

The encoding field is used to specify the format and compression of the batch files. Currently only jsonl and gzip are supported, respectively.

manifest

The manifest field is used to specify the paths to the batch files. The paths are relative to the root directory specified in the batch_config storage configuration.

Batch configuration

When local storage is used, targets do no require special configuration to process BATCH messages.

Taps may be configured to specify a root storage root directory, file path prefix, and encoding for batch files using a configuration like the below:

In config.json:

{
  // ...
  "batch_config": {
    "encoding": {
      "format": "jsonl",
      "compression": "gzip",
    },
    "storage": {
      "root": "file://tests/core/resources",
      "prefix": "test-batch-",
    }
  }
}

Custom batch file creation and processing

Tap side

Taps can optionally customize the batch file creation by implementing the get_batches. This method should return a tuple of an encoding and a list of batch files:

class MyStream(Stream):
    def get_batches(self, records):
        return (
            ParquetEncoding(compression="snappy"),
            [
                "s3://my-bucket/my-batch-file-1.parquet",
                "s3://my-bucket/my-batch-file-2.parquet",
            ]
        )

Target side

Targets can optionally customize the batch file processing by implementing the process_batch_files.

class MySink(Sink):
    def process_batch_files(self, encoding, storage, files):
        # process the batch files