embulk

Embulk configuration file format

Embulk uses a YAML file to define a bulk data loading. Here is an example of the file:

in:
  type: file
  path_prefix: ./mydata/csv/
  decoders:
  - {type: gzip}
  parser:
    charset: UTF-8
    newline: CRLF
    type: csv
    delimiter: ','
    quote: '"'
    escape: '"'
    null_string: 'NULL'
    skip_header_lines: 1
    columns:
    - {name: id, type: long}
    - {name: account, type: long}
    - {name: time, type: timestamp, format: '%Y-%m-%d %H:%M:%S'}
    - {name: purchase, type: timestamp, format: '%Y%m%d'}
    - {name: comment, type: string}
filters:
  - type: speedometer
    speed_limit: 250000
out:
  type: stdout

A configuration file consists of following sections:

• in: Input plugin options. An input plugin is either record-based (MySQL, DynamoDB, etc) or file-based (S3, HTTP, etc).

  • parser: If the input is file-based, parser plugin parses a file format (built-in csv, json, etc).

  • decoder: If the input is file-based, decoder plugin decodes compression or encryption (built-in gzip, bzip2, zip, tar.gz, etc).

• out: Output plugin options. An output plugin is either record-based (Oracle, Elasticsearch, etc) or file-based (Google Cloud Storage, Command, etc)

  • formatter: If the output is file-based, formatter plugin formats a file format (such as built-in csv, jsonl)

  • encoder: If the output is file-based, encoder plugin encodes compression or encryption (such as built-in gzip or bzip2)

• filters: Filter plugins options (optional).

• exec: Executor plugin options. An executor plugin control parallel processing (such as built-in thread executor)

In many cases, what you need to write is in:, out: and formatter: sections only because guess command guesses parser: and decoder: options for you. See also the Quick Start.

Using variables

You can embed environment variables in configuration file using Liquid template engine (This is experimental feature. Behavior might change or be removed in future releases).

To use template engine, configuration file name must end with .yml.liquid.

Environment variables are set to env variable.

in:
  type: file
  path_prefix: 
  decoders:
  - {type: gzip}
  parser:
    ...
out:
  type: postgresql
  host: 
  port: 
  user: 
  password: ""
  database: embulk_load
  mode: insert
  table: 

Including files

Configuration file can include another configuration file. To use it, configuration file name must end with .yml.liquid.

File will be searched from the relative path of the input configuration file. And file name will be _<name>.yml.liquid. For example, if you add {% include 'subdir/inc' %} tag to myconfig/config.yml.liquid file, it includes myconfig/subdir/_inc.yml.liquid file.

# config.yml.liquid
{% include 'in_mysql' %}
out:
  type: stdout

# _in_mysql.yml.liquid
in:
  type: mysql

With above 2 files, actual configuration file will be:

# $ embulk run config.yml.liquid
in:
  type: mysql
out:
  type: stdout

Local file input plugin

The file input plugin reads files from local file system.

Options

The path_prefix option is required. If you have files as following, you may set path_prefix: /path/to/files/sample_:

.
`-- path
    `-- to
        `-- files
            |-- sample_01.csv   -> read
            |-- sample_02.csv   -> read
            |-- sample_03.csv   -> read
            |-- sample_04.csv   -> read

The last_path option is used to skip files older than or same with the file in dictionary order. For example, if you set last_path: /path/to/files/sample_02.csv, Embulk reads following files:

.
`-- path
    `-- to
        `-- files
            |-- sample_01.csv   -> skip
            |-- sample_02.csv   -> skip
            |-- sample_03.csv   -> read
            |-- sample_04.csv   -> read

Example

in:
  type: file
  path_prefix: /path/to/files/sample_
  last_path: /path/to/files/sample_02.csv
  parser:
    ...

In most of cases, you’ll use guess to configure the parser and decoders. See also Quick Start.

CSV parser plugin

The csv parser plugin parses CSV and TSV files.

Options

The quotes_in_quoted_fields option specifies how to deal with irregular non-escaped stray quote characters.

The columns option declares the list of columns. This CSV parser plugin ignores the header line.

List of types:

The null_string option converts certain values to NULL. Values will be converted as following:

You can use guess to automatically generate the column settings. See also Quick Start.

Example

in:
  ...
  parser:
    type: csv
    charset: UTF-8
    newline: CRLF
    delimiter: "\t"
    quote: '"'
    escape: '"'
    null_string: 'NULL'
    skip_header_lines: 1
    comment_line_marker: '#'
    columns:
    - {name: id, type: long}
    - {name: account, type: long}
    - {name: time, type: timestamp, format: '%Y-%m-%d %H:%M:%S'}
    - {name: purchase, type: timestamp, format: '%Y%m%d'}
    - {name: comment, type: string}

JSON parser plugin

The json parser plugin parses a JSON file that contains a sequence of JSON objects. Example:

{"time":1455829282,"ip":"93.184.216.34","name":"frsyuki"}
{"time":1455829282,"ip":"172.36.8.109", "name":"sadayuki"}
{"time":1455829284,"ip":"example.com","name":"Treasure Data"}
{"time":1455829282,"ip":"10.98.43.1","name":"MessagePack"}

json parser plugin outputs a single record named “record” (type is json).

Options

if you set invalid_string_escapes and appear invalid JSON string (such as \a), it makes following the action.

(*1): Throwing an exception.

The columns option declares the list of columns, and the way how to extract JSON values into Embulk columns.

Example

in:
  parser:
    type: json
    columns:
    - {name: time, type: timestamp, format: "%s"}
    - {name: ip, type: string}
    - {name: name, type: string}

Gzip decoder plugin

The gzip decoder plugin decompresses gzip files before input plugins read them.

Options

This plugin doesn’t have any options.

Example

in:
  ...
  decoders:
  - {type: gzip}

BZip2 decoder plugin

The bzip2 decoder plugin decompresses bzip2 files before input plugins read them.

Options

This plugin doesn’t have any options.

Example

in:
  ...
  decoders:
  - {type: bzip2}

File output plugin

The file output plugin writes records to local file system.

Options

For example, if you set path_prefix: /path/to/output/sample_, sequence_format: "%03d.%02d.", and file_ext: csv, name of the output files will be as following:

.
`-- path
    `-- to
        `-- output
            |-- sample_01.000.csv
            |-- sample_02.000.csv
            |-- sample_03.000.csv
            |-- sample_04.000.csv

sequence_format formats task index and sequence number in a task.

Example

out:
  type: file
  path_prefix: /path/to/output/sample_
  file_ext: csv
  formatter:
    ...

CSV formatter plugin

The csv formatter plugin formats records using CSV or TSV format.

Options

(*1): if quote_policy is NONE, quote option is ignored, and default escape is \.

The quote_policy option is used to determine field type to quote.

The column_options option is a map whose keys are name of columns, and values are configuration with following parameters:

Example

out:
  ...
  formatter:
    type: csv
    delimiter: "\t"
    newline: CRLF
    newline_in_field: LF
    charset: UTF-8
    quote_policy: MINIMAL
    quote: '"'
    escape: "\\"
    null_string: "\\N"
    default_timezone: 'UTC'
    column_options:
      mycol1: {format: '%Y-%m-%d %H:%M:%S'}
      mycol2: {format: '%Y-%m-%d %H:%M:%S', timezone: 'America/Los_Angeles'}

Gzip encoder plugin

The gzip encoder plugin compresses output files using gzip.

Options

Example

out:
  ...
  encoders:
  - type: gzip
    level: 1

BZip2 encoder plugin

The bzip2 encoder plugin compresses output files using bzip2.

Options

Example

out:
  ...
  encoders:
  - type: bzip2
    level: 6

Rename filter plugin

The rename filter plugin changes column names. This plugin has no impact on performance.

Options

Renaming rules

The rules is an array of rules as below applied top-down for all the columns.

Renaming rule: character_types

The rule character_types replaces restricted characters.

Example

# This configuration replaces characters into "_" except for "_", lower-case alphabets, and digits.
filters:
  ...
  - type: rename
    rules:
    - rule: character_types
      pass_characters: "_"
      pass_types: [ "a-z", "0-9" ]

Renaming rule: first_character_types

The rule first_character_types prefixes or replaces a restricted character at the beginning.

Example

# This configuration prefixes a column name with "_" unless the name starts from "_" or a lower-case alphabet.
filters:
  ...
  - type: rename
    rules:
    - rule: first_character_types
      pass_characters: "_"
      pass_types: [ "a-z" ]
      prefix: "_"

Renaming rule: lower_to_upper

The rule lower_to_upper converts lower-case alphabets to upper-case.

Example

# This configuration converts all lower-case alphabets to upper-case.
filters:
  ...
  - type: rename
    rules:
    - rule: lower_to_upper

Renaming rule: regex_replace

The rule regex_replace replaces column names based on a regular expression.

Example

# This configuration replaces all patterns
filters:
  ...
  - type: rename
    rules:
    - rule: regex_replace
      match: "([0-9]+)_dollars"
      replace: "USD$1"

Renaming rule: truncate

The rule truncate truncates column names.

Example

# This configuration drops all characters after the 20th character.
filters:
  ...
  - type: rename
    rules:
    - rule: truncate
      max_length: 20

Renaming rule: upper_to_lower

The rule upper_to_lower converts upper-case alphabets to lower-case.

Example

# This configuration converts all upper-case alphabets to lower-case.
filters:
  ...
  - type: rename
    rules:
    - rule: upper_to_lower

Renaming rule: unique_number_suffix

The rule unique_number_suffix makes column names unique in the schema by suffixing numbers.

Example

# This configuration suffixes numbers to duplicative column names. (Ex. ["column", "column", "column"] goes to ["column", "column_2", "column_3"].)
filters:
  ...
  - type: rename
    rules:
    - rule: unique_number_suffix

Example of renaming rules

filters:
  ...
  - type: rename
    rules:
    - rule: upper_to_lower        # All upper-case are converted to lower-case.
    - rule: character_types       # Only lower-case, digits and "_" are allowed. (No upper-case by the rule ahove.)
      pass_types: [ "a-z", "0-9" ]
      pass_characters: "_"
    - rule: unique_number_suffix  # Ensure all column names are unique.

Columns: not recommended

columns is not recommended to use anymore. Consider using rules instead.

filters:
  ...
  - type: rename
    columns:
      my_existing_column1: new_column1
      my_existing_column2: new_column2

Remove columns filter plugin

The remove_columns filter plugin removes columns from schema.

Options

remove: and keep: options are not multi-select.

Example

# This configuration removes "_c0" and "_c1" named columns from schema.
filters:
  ...
  - type: remove_columns
    remove: ["_c0", "_c1"]

Local executor plugin

The local executor plugin runs tasks using local threads. This is the only built-in executor plugin.

Options

The max_threads option controls maximum concurrency. Setting smaller number here is useful if too many threads make the destination or source storage overloaded. Setting larger number here is useful if CPU utilization is too low due to high latency.

The min_output_tasks option enables “page scattering”. The feature is enabled if number of input tasks is less than min_output_tasks. It uses multiple filter & output threads for each input task so that one input task can use multiple threads. Setting larger number here is useful if embulk doesn’t use multi-threading with enough concurrency due to too few number of input tasks. Setting 1 here disables page scattering completely.

Example

exec:
  max_threads: 8         # run at most 8 tasks concurrently
  min_output_tasks: 1    # disable page scattering
in:
  type: ...
  ...
out:
  type: ...
  ...

Guess executor

The guess executor is called by guess command. It executes default guess plugins in a sequential order and suggests Embulk config by appropriate guess plugin. The default guess plugins and the order are gzip, 'bzip2, json and csv.

Options

The guess_plugins option includes specified guess plugin in the bottom of the list of default guess plugins.

The exclude_guess_plugins option exclude specified guess plugins from the list of default guess plugins that the guess executor uses.

The guess_sample_buffer_bytes option controls the bytes of sample buffer that GuessExecutor tries to read from specified input source.

This example shows how to use csv_all_strings guess plugin, which suggests column types within CSV files as string types. It needs to be explicitly specified by users when it’s used instead of csv guess plugin because the plugin is not included in default guess plugins. We also can exclude default csv guess plugin.

Example

exec:
  guess_plugins: ['csv_all_strings']
  exclude_guess_plugins: ['csv']
in:
  type: ...
  ...
out:
  type: ...
  ...

Preview executor

The preview executor is called by preview command. It tries to read sample buffer from a specified input source and writes them to Page objects. preview outputs the Page objects to console.

Options

The preview_sample_buffer_bytes option controls the bytes of sample buffer that PreviewExecutor tries to read from specified input source.

This example shows how to change the bytes of sample buffer.

Example

exec:
  preview_sample_buffer_bytes: 65536 # 64KB
in:
  type: ...
  ...
out:
  type: ...
  ...