Module: Fluent::BigQueryOutput::LoadImplementation

Defined in:

lib/fluent/plugin/out_bigquery.rb

Instance Method Summary

• #_write(chunk, table_id) ⇒ Object
• #format(tag, time, record) ⇒ Object

Instance Method Details

#_write(chunk, table_id) ⇒ Object

# File 'lib/fluent/plugin/out_bigquery.rb', line 456

def _write(chunk, table_id)
  res = nil
  job_id = nil

  create_upload_source(chunk) do |upload_source|
    if @prevent_duplicate_load
      job_id = create_job_id(upload_source.path, @dataset, @table, @fields.to_a, @max_bad_records, @ignore_unknown_values)
    end
    configuration = {
      configuration: {
        load: {
          destination_table: {
            project_id: @project,
            dataset_id: @dataset,
            table_id: table_id,
          },
          schema: {
            fields: @fields.to_a,
          },
          write_disposition: "WRITE_APPEND",
          source_format: "NEWLINE_DELIMITED_JSON",
          ignore_unknown_values: @ignore_unknown_values,
          max_bad_records: @max_bad_records,
        }
      }
    }
    configuration.merge!({job_reference: {project_id: @project, job_id: job_id}}) if job_id
    res = client.insert_job(@project, configuration, {upload_source: upload_source, content_type: "application/octet-stream"})
  end

  wait_load(res.job_reference.job_id)
rescue Google::Apis::ServerError, Google::Apis::ClientError, Google::Apis::AuthorizationError => e
  # api_error? -> client cache clear
  @cached_client = nil

  reason = e.respond_to?(:reason) ? e.reason : nil
  log.error "job.insert API", project_id: @project, dataset: @dataset, table: table_id, code: e.status_code, message: e.message, reason: reason

  raise "failed to insert into bigquery, retry" if reason == "backendError" # backendError is retryable. TODO: error class
  return wait_load(job_id) if e.status_code == 409 && e.message =~ /Job/ # duplicate load job

  # other error handling
  if @secondary
    flush_secondary(@secondary)
  end
end

#format(tag, time, record) ⇒ Object

# File 'lib/fluent/plugin/out_bigquery.rb', line 443

def format(tag, time, record)
  buf = ''

  if @replace_record_key
    record = replace_record_key(record)
  end
  row = @fields.format(@add_time_field.call(record, time))
  unless row.empty?
    buf << MultiJson.dump(row) + "\n"
  end
  buf
end