Class: Fluent::BigQueryOutput::RecordSchema
- Inherits:
-
FieldSchema
- Object
- FieldSchema
- Fluent::BigQueryOutput::RecordSchema
- 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 Attribute Summary
Attributes inherited from FieldSchema
Instance Method Summary collapse
- #[](name) ⇒ Object
- #format_one(record) ⇒ Object
-
#initialize(name, mode = :nullable) ⇒ RecordSchema
constructor
A new instance of RecordSchema.
- #load_schema(schema, allow_overwrite = true) ⇒ Object
- #register_field(name, type) ⇒ Object
- #to_a ⇒ Object
- #to_h ⇒ Object
- #type ⇒ Object
Methods inherited from FieldSchema
Constructor Details
#initialize(name, mode = :nullable) ⇒ RecordSchema
Returns a new instance of RecordSchema.
622 623 624 625 |
# File 'lib/fluent/plugin/out_bigquery.rb', line 622 def initialize(name, mode = :nullable) super(name, mode) @fields = {} end |
Instance Method Details
#[](name) ⇒ Object
631 632 633 |
# File 'lib/fluent/plugin/out_bigquery.rb', line 631 def [](name) @fields[name] end |
#format_one(record) ⇒ Object
688 689 690 691 692 693 694 695 696 697 |
# File 'lib/fluent/plugin/out_bigquery.rb', line 688 def format_one(record) out = {} @fields.each do |key, schema| value = record[key] formatted = schema.format(value) next if formatted.nil? # field does not exists, or null value out[key] = formatted end out end |
#load_schema(schema, allow_overwrite = true) ⇒ Object
650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 |
# File 'lib/fluent/plugin/out_bigquery.rb', line 650 def load_schema(schema, allow_overwrite=true) schema.each do |field| raise ConfigError, 'field must have type' unless field.key?('type') name = field['name'] mode = (field['mode'] || 'nullable').downcase.to_sym type = field['type'].downcase.to_sym field_schema_class = FIELD_TYPES[type] raise ConfigError, "Invalid field type: #{field['type']}" unless field_schema_class next if @fields.key?(name) and !allow_overwrite field_schema = field_schema_class.new(name, mode) @fields[name] = field_schema if type == :record raise ConfigError, "record field must have fields" unless field.key?('fields') field_schema.load_schema(field['fields'], allow_overwrite) end end end |
#register_field(name, type) ⇒ Object
672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 |
# File 'lib/fluent/plugin/out_bigquery.rb', line 672 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(name) end end |
#to_a ⇒ Object
635 636 637 638 639 |
# File 'lib/fluent/plugin/out_bigquery.rb', line 635 def to_a @fields.map do |_, field_schema| field_schema.to_h end end |
#to_h ⇒ Object
641 642 643 644 645 646 647 648 |
# File 'lib/fluent/plugin/out_bigquery.rb', line 641 def to_h { :name => name, :type => type.to_s.upcase, :mode => mode.to_s.upcase, :fields => self.to_a, } end |
#type ⇒ Object
627 628 629 |
# File 'lib/fluent/plugin/out_bigquery.rb', line 627 def type :record end |