Module: Fluent::Plugin::BigQueryOutput::InsertImplementation
- Defined in:
- lib/fluent/plugin/out_bigquery.rb
Instance Method Summary collapse
- #_write(chunk, table_format) ⇒ Object
- #insert(project, dataset, table_id, rows, schema, template_suffix) ⇒ Object
Instance Method Details
#_write(chunk, table_format) ⇒ Object
383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 |
# File 'lib/fluent/plugin/out_bigquery.rb', line 383 def _write(chunk, table_format) now = Time.now.utc.strftime("%Y-%m-%d %H:%M:%S.%6N") if @add_insert_timestamp rows = chunk.open do |io| io.map do |line| record = MultiJson.load(line) record[@add_insert_timestamp] = now if @add_insert_timestamp row = {"json" => record} row["insert_id"] = @get_insert_id.call(record) if @get_insert_id Fluent::BigQuery::Helper.deep_symbolize_keys(row) end end project = extract_placeholders(@project, chunk.) dataset = extract_placeholders(@dataset, chunk.) table_id = extract_placeholders(table_format, chunk.) template_suffix = @template_suffix ? extract_placeholders(@template_suffix, chunk.) : nil schema = get_schema(project, dataset, chunk.) insert(project, dataset, table_id, rows, schema, template_suffix) end |
#insert(project, dataset, table_id, rows, schema, template_suffix) ⇒ Object
405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 |
# File 'lib/fluent/plugin/out_bigquery.rb', line 405 def insert(project, dataset, table_id, rows, schema, template_suffix) writer.insert_rows(project, dataset, table_id, rows, template_suffix: template_suffix) rescue Fluent::BigQuery::Error => e if @auto_create_table && e.status_code == 404 && /Not Found: Table/i =~ e. # Table Not Found: Auto Create Table writer.create_table(project, dataset, table_id, schema) raise "table created. send rows next time." end raise if e.retryable? if @secondary # TODO: find better way @retry = retry_state_create( :output_retries, @buffer_config.retry_type, @buffer_config.retry_wait, @buffer_config.retry_timeout, forever: false, max_steps: @buffer_config.retry_max_times, backoff_base: @buffer_config.retry_exponential_backoff_base, max_interval: @buffer_config.retry_max_interval, secondary: true, secondary_threshold: Float::EPSILON, randomize: @buffer_config.retry_randomize ) else @retry = retry_state_create( :output_retries, @buffer_config.retry_type, @buffer_config.retry_wait, @buffer_config.retry_timeout, forever: false, max_steps: 0, backoff_base: @buffer_config.retry_exponential_backoff_base, max_interval: @buffer_config.retry_max_interval, randomize: @buffer_config.retry_randomize ) end raise end |