Class: Fluent::BigQueryOutput::RecordSchema

Inherits:
FieldSchema
  • Object
show all
Defined in:
lib/fluent/plugin/out_bigquery.rb

Constant Summary collapse

FIELD_TYPES = 
{
  :string => StringFieldSchema,
  :integer => IntegerFieldSchema,
  :float => FloatFieldSchema,
  :boolean => BooleanFieldSchema,
  :timestamp => TimestampFieldSchema,
  :record => RecordSchema
}.freeze

Instance Method Summary collapse

Constructor Details

#initialize ⇒ RecordSchema

Returns a new instance of RecordSchema.



385
386
387
 
# File 'lib/fluent/plugin/out_bigquery.rb', line 385

def initialize
  @fields = {}
end

Instance Method Details

#[](name) ⇒ Object 



393
394
395
 
# File 'lib/fluent/plugin/out_bigquery.rb', line 393

def [](name)
  @fields[name]
end

#format(record) ⇒ Object 



430
431
432
433
434
435
436
437
438
 
# File 'lib/fluent/plugin/out_bigquery.rb', line 430

def format(record)
  out = {}
  @fields.each do |key, schema|
    value = record[key]
    next if value.nil? # field does not exists, or null value
    out[key] = schema.format(value)
  end
  out
end

#load_schema(schema) ⇒ Object 



397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
 
# File 'lib/fluent/plugin/out_bigquery.rb', line 397

def load_schema(schema)
  schema.each do |field|
    raise ConfigError, 'field must have type' unless field.key?('type')

    type = field['type'].downcase.to_sym
    field_schema_class = FIELD_TYPES[type]
    raise ConfigError, "Invalid field type: #{field['type']}" unless field_schema_class

    field_schema = field_schema_class.new
    @fields[field['name']] = field_schema
    if type == :record
      raise ConfigError, "record field must have fields" unless field.key?('fields')
      field_schema.load_schema(field['fields'])
    end
  end
end

#register_field(name, type) ⇒ Object 



414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
 
# File 'lib/fluent/plugin/out_bigquery.rb', line 414

def register_field(name, type)
  if @fields.key?(name) and @fields[name].type != :timestamp
    raise ConfigError, "field #{name} is registered twice"
  end
  if name[/\./]
    recordname = $`
    fieldname = $'
    register_record_field(recordname)
    @fields[recordname].register_field(fieldname, type)
  else
    schema = FIELD_TYPES[type]
    raise ConfigError, "[Bug] Invalid field type #{type}" unless schema
    @fields[name] = schema.new
  end
end

#type ⇒ Object 



389
390
391
 
# File 'lib/fluent/plugin/out_bigquery.rb', line 389

def type
  :record
end