Class: Fluent::Plugin::OpenlineageParser

Inherits:
JSONParser
  • Object
show all
Defined in:
lib/fluent/plugin/parser_openlineage.rb

Defined Under Namespace

Classes: BadRequestError

Constant Summary collapse

DEFAULT_SPEC_DIRECTORY = 
"/etc/spec"

Instance Method Summary collapse

Instance Method Details

#configure(conf) ⇒ Object 



13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
 
# File 'lib/fluent/plugin/parser_openlineage.rb', line 13

def configure(conf)
  if conf.has_key?('spec_directory')
    @spec_directory = conf['spec_directory']
  else
    @spec_directory = DEFAULT_SPEC_DIRECTORY
  end
  if (not @spec_directory.end_with?("/"))
    @spec_directory += "/"
  end
  @validate_input_dataset_facets = conf.fetch('validate_input_dataset_facets', false)
  @validate_output_dataset_facets = conf.fetch('validate_output_dataset_facets', false)
  @validate_dataset_facets = conf.fetch('validate_dataset_facets', false)
  @validate_run_facets = conf.fetch('validate_run_facets', true)
  @validate_job_facets = conf.fetch('validate_job_facets', true)
  @schema = load_schema()
  @validator = RustyJSONSchema.build(@schema)
  super
end

#parse(text) ⇒ Object

docs.fluentd.org/plugin-development/api-plugin-parser



33
34
35
36
37
38
39
 
# File 'lib/fluent/plugin/parser_openlineage.rb', line 33

def parse(text)
  # parse JSON with default JSONParser
  super(text) { | time, json |
  validate_openlineage(json)
  yield time, json
  }
end