Module: Fluent::BigQueryOutput::LoadImplementation

Defined in:

lib/fluent/plugin/out_bigquery.rb

Instance Method Summary

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

Instance Method Details

#_write(chunk, table_id, template_suffix) ⇒ Object

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

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

  create_upload_source(chunk) do |upload_source|
    configuration, job_id = load_configuration(table_id, template_suffix, upload_source)
    res = client.insert_job(
      @project,
      configuration,
      {
        upload_source: upload_source,
        content_type: "application/octet-stream",
        options: {
          timeout_sec: @request_timeout_sec,
          open_timeout_sec: @request_open_timeout_sec,
        }
      }
    )
  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

  return wait_load(job_id) if job_id && e.status_code == 409 && e.message =~ /Job/ # duplicate load job

  if RETRYABLE_ERROR_REASON.include?(reason) || e.is_a?(Google::Apis::ServerError)
    raise "failed to insert into bigquery, retry" # TODO: error class
  elsif @secondary
    flush_secondary(@secondary)
  end
end

#format(tag, time, record) ⇒ Object

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

def format(tag, time, record)
  fetch_schema if @template_suffix

  if @replace_record_key
    record = replace_record_key(record)
  end

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