ElasticGraph::JSONIngestion
Provides JSON Schema ingestion support for ElasticGraph.
This gem provides the schema-definition extension that generates JSON Schema artifacts for indexing
events and validates JSON-ingestion-specific schema options. Generated ElasticGraph projects install
and enable it by default. Applications that wire schema-definition tasks manually enable it by adding
ElasticGraph::JSONIngestion::SchemaDefinition::APIExtension to their schema-definition extension modules.
Schema Definition APIs
Use schema.json_schema_version to identify the current JSON schema artifact. Every change that affects
the JSON schema should increment this version so publishers and indexers can safely evolve independently.
During early prototyping, schema.enforce_json_schema_version false can disable version-change enforcement.
For production applications, leave enforcement enabled so schema artifact changes require an intentional
version bump.
diff --git a/config/schema.rb b/config/schema.rb
index 015c5fa..b8eeaef 100644
--- a/config/schema.rb
+++ b/config/schema.rb
@@ -2,3 +2,3 @@ ElasticGraph.define_schema do |schema|
# ElasticGraph will tell you when you need to bump this.
- schema.json_schema_version 1
+ schema.json_schema_version 2
# in config/schema/json_schema_enforcement.rb
ElasticGraph.define_schema do |schema|
schema.enforce_json_schema_version false
end
Use schema.json_schema_strictness to configure whether indexing events may omit nullable fields or include
extra fields. We recommend enabling at most one of these options, because enabling both can hide misspelled
event fields.
# in config/schema/json_schema_strictness.rb
ElasticGraph.define_schema do |schema|
schema.json_schema_strictness allow_omitted_fields: true, allow_extra_fields: false
end
Custom scalar types must declare how they are represented in JSON Schema:
# in config/schema/url.rb
ElasticGraph.define_schema do |schema|
schema.scalar_type "URL" do |t|
t.mapping type: "keyword"
t.json_schema type: "string", format: "uri"
end
end
Fields and object/interface types can add JSON Schema validations. Use field-level validations sparingly: they run while indexing events, so violations can send otherwise valid source-system data to the dead letter queue. They are best reserved for constraints that ElasticGraph needs in order to index correctly.
# in config/schema/card.rb
ElasticGraph.define_schema do |schema|
schema.object_type "Card" do |t|
t.field "id", "ID!"
t.field "expYear", "Int" do |f|
f.json_schema minimum: 2000, maximum: 2099
end
t.field "expMonth", "Int" do |f|
f.json_schema minimum: 1, maximum: 12
end
t.index "cards"
end
end
On fields, nullable: false disallows null in indexing events while keeping the GraphQL field nullable:
# in config/schema/widget.rb
ElasticGraph.define_schema do |schema|
schema.object_type "Widget" do |t|
t.field "id", "ID!"
t.field "name", "String" do |f|
f.json_schema nullable: false
end
t.index "widgets"
end
end
Indexing Support
Beyond schema definition, this gem provides an adapter used by elasticgraph-indexer to ingest JSON events. The
adapter validates each event envelope against the JSON schema identified by the event's json_schema_version,
builds an ElasticGraph::Indexer::Event for each valid envelope, validates each record, and prepares records for
indexing using that version's view of the schema.
JSON ingestion is enabled automatically for schemas defined with this gem's SchemaDefinition::APIExtension;
no indexer configuration is needed.
This gem also provides the be_a_valid_elastic_graph_event RSpec matcher (via
require "elastic_graph/json_ingestion/spec_support/event_matcher") for testing that publisher events
conform to your schema.
Processing JSON Lines Payloads
ElasticGraph::JSONIngestion::Indexer wraps the format-neutral ElasticGraph::Indexer with JSON Lines decoding:
require "elastic_graph/json_ingestion/indexer"
indexer = ElasticGraph::JSONIngestion::Indexer.from_yaml_file("config/settings/local.yaml")
json_lines_payload = $stdin.read
indexer.process(json_lines_payload) unless json_lines_payload.empty?
Use #process_returning_failures when individual failures must be handled by the caller. It accepts decoded
events, so a transport that must add metadata or combine several payloads into one bulk operation can call
#decode first, adjust the decoded events, and then process them:
require "elastic_graph/json_ingestion/indexer"
indexer = ElasticGraph::JSONIngestion::Indexer.from_yaml_file("config/settings/local.yaml")
# One JSON Lines payload per blank-line-delimited chunk.
payloads = $stdin.read.split("\n\n")
decoded_events = payloads.flat_map { |payload| indexer.decode(payload) }
unless decoded_events.empty?
failures = indexer.process_returning_failures(decoded_events)
abort "#{failures.size} of #{decoded_events.size} event(s) failed." if failures.any?
end
The wrapper accepts an existing ElasticGraph::Indexer, so independent JSON and protobuf wrappers can share the
same format-neutral indexer and its datastore clients:
require "elastic_graph/json_ingestion/indexer"
base_indexer = ElasticGraph::Indexer.from_yaml_file("config/settings/local.yaml")
json_indexer = ElasticGraph::JSONIngestion::Indexer.new(base_indexer)