embulk-output-bigquery

Embulk output plugin to load/insert data into Google BigQuery using direct insert

load data into Google BigQuery as batch jobs for big amount of data https://developers.google.com/bigquery/loading-data-into-bigquery

• Plugin type: output
• Resume supported: no
• Cleanup supported: no
• Dynamic table creating: yes

• insert data over streaming inserts
  • for continuous real-time insertions
  • Please use other product, like fluent-plugin-bigquery
  • https://developers.google.com/bigquery/streaming-data-into-bigquery#usecases

Current version of this plugin supports Google API with Service Account Authentication, but does not support OAuth flow for installed applications.

Client or request options

Options for intermediate local files

Options for intermediate tables on BigQuery

source_format is also used to determine formatter (csv or jsonl).

Following options are same as bq command-line tools or BigQuery job's property.

out:
  type: bigquery
  mode: append
  auth_method: service_account
  json_keyfile: /path/to/json_keyfile.json
  project: your-project-000
  dataset: your_dataset_name
  table: your_table_name
  compression: GZIP
  source_format: NEWLINE_DELIMITED_JSON

The geographic location of the dataset. Required except for US and EU.

GCS bucket should be in same region when you use gcs_bucket.

See also Dataset Locations | BigQuery | Google Cloud

5 modes are provided.

1. Load to temporary table (Create and WRITE_APPEND in parallel)
2. Copy temporary table to destination table (or partition). (WRITE_APPEND)

1. Insert data into existing table (or partition) directly. (WRITE_APPEND in parallel)

This is not transactional, i.e., if fails, the target table could have some rows inserted.

1. Load to temporary table (Create and WRITE_APPEND in parallel)
2. Copy temporary table to destination table (or partition). (WRITE_TRUNCATE)

is_skip_job_result_check must be false when replace mode

NOTE: BigQuery does not support replacing (actually, copying into) a non-partitioned table with a paritioned table atomically. You must once delete the non-partitioned table, otherwise, you get Incompatible table partitioning specification when copying to the column partitioned table error.

1. Load to temporary table (Create and WRITE_APPEND in parallel)
2. Copy destination table (or partition) to backup table (or partition). (dataset_old, table_old)
3. Copy temporary table to destination table (or partition). (WRITE_TRUNCATE)

is_skip_job_result_check must be false when replace_backup mode.

1. Delete destination table (or partition), if it exists.
2. Load to destination table (or partition).

There are four authentication methods

1. service_account (or json_key for backward compatibility)
2. authorized_user
3. compute_engine
4. application_default

Use GCP service account credentials. You first need to create a service account, download its json key and deploy the key with embulk.

out:
  type: bigquery
  auth_method: service_account
  json_keyfile: /path/to/json_keyfile.json

You can also embed contents of json_keyfile at config.yml.

out:
  type: bigquery
  auth_method: service_account
  json_keyfile:
    content: |
      {
          "private_key_id": "123456789",
          "private_key": "-----BEGIN PRIVATE KEY-----\nABCDEF",
          "client_email": "..."
      }

Use Google user credentials. You can get your credentials at ~/.config/gcloud/application_default_credentials.json by running gcloud auth login.

out:
  type: bigquery
  auth_method: authorized_user
  json_keyfile: /path/to/credentials.json

You can also embed contents of json_keyfile at config.yml.

out:
  type: bigquery
  auth_method: authorized_user
  json_keyfile:
    content: |
      {
        "client_id":"xxxxxxxxxxx.apps.googleusercontent.com",
        "client_secret":"xxxxxxxxxxx",
        "refresh_token":"xxxxxxxxxxx",
        "type":"authorized_user"
      }

On the other hand, you don't need to explicitly create a service account for embulk when you run embulk in Google Compute Engine. In this third authentication method, you need to add the API scope "https://www.googleapis.com/auth/bigquery" to the scope list of your Compute Engine VM instance, then you can configure embulk like this.

out:
  type: bigquery
  auth_method: compute_engine

Use Application Default Credentials (ADC). ADC is a strategy to locate Google Cloud Service Account credentials.

1. ADC checks to see if the environment variable GOOGLE_APPLICATION_CREDENTIALS is set. If the variable is set, ADC uses the service account file that the variable points to.
2. ADC checks to see if ~/.config/gcloud/application_default_credentials.json is located. This file is created by running gcloud auth application-default login.
3. Use the default service account for credentials if the application running on Compute Engine, App Engine, Kubernetes Engine, Cloud Functions or Cloud Run.

See https://cloud.google.com/docs/authentication/production for details.

out:
  type: bigquery
  auth_method: application_default

table and option accept Time#strftime format to construct table ids. Table ids are formatted at runtime using the local time of the embulk server.

For example, with the configuration below, data is inserted into tables table_20150503, table_20150504 and so on.

out:
  type: bigquery
  table: table_%Y%m%d

There are 3 ways to set schema.

Please set file path of schema.json.

out:
  type: bigquery
  auto_create_table: true
  table: table_%Y%m%d
  schema_file: /path/to/schema.json

Plugin will try to read schema from existing table and use it as schema template.

out:
  type: bigquery
  auto_create_table: true
  table: table_%Y%m%d
  template_table: existing_table_name

Plugin will try to guess BigQuery schema from Embulk schema. It is also configurable with column_options. See Column Options.

Column options are used to aid guessing BigQuery schema, or to define conversion of values:

• column_options: advanced: an array of options for columns
  • name: column name
  • type: BigQuery type such as BOOLEAN, INTEGER, FLOAT, STRING, TIMESTAMP, DATETIME, DATE, and RECORD. See belows for supported conversion type.
    • boolean: BOOLEAN, STRING (default: BOOLEAN)
    • long: BOOLEAN, INTEGER, FLOAT, STRING, TIMESTAMP (default: INTEGER)
    • double: INTEGER, FLOAT, STRING, TIMESTAMP (default: FLOAT)
    • string: BOOLEAN, INTEGER, FLOAT, STRING, TIME, TIMESTAMP, DATETIME, DATE, RECORD (default: STRING)
    • timestamp: INTEGER, FLOAT, STRING, TIME, TIMESTAMP, DATETIME, DATE (default: TIMESTAMP)
    • json: STRING, RECORD (default: STRING)
  • mode: BigQuery mode such as NULLABLE, REQUIRED, and REPEATED (string, default: NULLABLE)
  • fields: Describes the nested schema fields if the type property is set to RECORD. Please note that this is required for RECORD column.
  • description: description (string, default is None).
  • timestamp_format: timestamp format to convert into/from timestamp (string, default is default_timestamp_format)
  • timezone: timezone to convert into/from timestamp, date (string, default is default_timezone).
• default_timestamp_format: default timestamp format for column_options (string, default is "%Y-%m-%d %H:%M:%S.%6N")
• default_timezone: default timezone for column_options (string, default is "UTC")

Example)

out:
  type: bigquery
  auto_create_table: true
  column_options:
    - {name: date, type: STRING, timestamp_format: %Y-%m-%d, timezone: "Asia/Tokyo"}
    - name: json_column
      type: RECORD
      fields:
        - {name: key1, type: STRING}
        - {name: key2, type: STRING}

NOTE: Type conversion is done in this jruby plugin, and could be slow. See Formatter Performance Issue to improve the performance.

embulk-output-bigquery supports formatting records into CSV or JSON (and also formatting timestamp column). However, this plugin is written in jruby, and jruby plugins are slower than java plugins generally.

Therefore, it is recommended to format records with filter plugins written in Java such as embulk-filter-to_json as:

filters:
  - type: to_json
    column: {name: payload, type: string}
    default_format: "%Y-%m-%d %H:%M:%S.%6N"
out:
  type: bigquery
  payload_column_index: 0 # or, payload_column: payload

Furtheremore, if your files are originally jsonl or csv files, you can even skip a parser with embulk-parser-none as:

in:
  type: file
  path_prefix: example/example.jsonl
  parser:
    type: none
    column_name: payload
out:
  type: bigquery
  payload_column_index: 0 # or, payload_column: payload

This is useful to reduce number of consumed jobs, which is limited by 100,000 jobs per project per day.

This plugin originally loads local files into BigQuery in parallel, that is, consumes a number of jobs, say 24 jobs on 24 CPU core machine for example (this depends on embulk parameters such as min_output_tasks and max_threads).

BigQuery supports loading multiple files from GCS with one job, therefore, uploading local files to GCS in parallel and then loading from GCS into BigQuery reduces number of consumed jobs to 1.

Using gcs_bucket option, such strategy is enabled. You may also use auto_create_gcs_bucket to create the specified GCS bucket automatically.

out:
  type: bigquery
  gcs_bucket: bucket_name
  auto_create_gcs_bucket: true

ToDo: Use https://cloud.google.com/storage/docs/streaming if google-api-ruby-client supports streaming transfers into GCS.

From 0.4.0, embulk-output-bigquery supports to load into partitioned table. See also Creating and Updating Date-Partitioned Tables.

To load into a partition, specify table parameter with a partition decorator as:

out:
  type: bigquery
  table: table_name$20160929

You may configure time_partitioning parameter together as:

out:
  type: bigquery
  table: table_name$20160929
  time_partitioning:
    type: DAY
    expiration_ms: 259200000

You can also create column-based partitioning table as:

out:
  type: bigquery
  mode: replace
  table: table_name
  time_partitioning:
    type: DAY
    field: timestamp

Note the time_partitioning.field should be top-level DATE or TIMESTAMP.

Use Tables: patch API to update the schema of the partitioned table, embulk-output-bigquery itself does not support it, though. Note that only adding a new column, and relaxing non-necessary columns to be NULLABLE are supported now. Deleting columns, and renaming columns are not supported.

MEMO: jobs#configuration.load.schemaUpdateOptions is available to update the schema of the desitination table as a side effect of the load job, but it is not available for copy job. Thus, it was not suitable for embulk-output-bigquery idempotence modes, append, replace, and replace_backup, sigh.

Prepare a json_keyfile at example/your-project-000.json, then

$ embulk bundle install --path vendor/bundle
$ embulk run -X page_size=1 -b . -l trace example/example.yml

Place your embulk with .jar extension:

$ curl -o embulk.jar --create-dirs -L "http://dl.embulk.org/embulk-latest.jar"
$ chmod a+x embulk.jar

Investigate JRUBY_VERSION and Bundler::VERSION included in the embulk.jar:

$ echo JRUBY_VERSION | ./embulk.jar irb
2019-08-10 00:59:11.866 +0900: Embulk v0.9.17
Switch to inspect mode.
JRUBY_VERSION
"X.X.X.X"

$ echo "require 'bundler'; Bundler::VERSION" | ./embulk.jar irb
2019-08-10 01:59:10.460 +0900: Embulk v0.9.17
Switch to inspect mode.
require 'bundler'; Bundler::VERSION
"Y.Y.Y"

Install the same version of jruby (change X.X.X.X to the version shown above) and bundler:

$ rbenv install jruby-X.X.X.X
$ rbenv local jruby-X.X.X.X
$ gem install bundler -v Y.Y.Y

Install dependencies (NOTE: Use bundler included in the embulk.jar, otherwise, gem 'embulk' is not found):

$ ./embulk.jar bundle install --path vendor/bundle

Run tests with env RUBYOPT="-r ./embulk.jar:

$ bundle exec env RUBYOPT="-r ./embulk.jar" rake test

To run tests which actually connects to BigQuery such as test/test_bigquery_client.rb, prepare a json_keyfile at example/your-project-000.json, then

$ bundle exec env RUBYOPT="-r ./embulk.jar" ruby test/test_bigquery_client.rb
$ bundle exec env RUBYOPT="-r ./embulk.jar" ruby test/test_example.rb

Change the version of gemspec, and write CHANGELOG.md. Then,

$ bundle exec rake release

CHANGELOG.md