A Struct holds named record fields with schema-defined type information. In Kafka Connect, it is the schema-aware counterpart to a schemaless Map: build the schema, create a Struct from it, and place field values into that typed container before producing a record. section 5.6.1 section 5.6.3
Use Struct when record shape is known and must travel with the data:
- Create or reuse the record schema, then instantiate a Struct with that schema. section 5.6.3
- Read values from the original Struct by field name and put them into a new Struct whose schema retains only selected fields. section 5.6.3
- Model nested ksqlDB objects with STRUCT<field type, ...>; dereference nested fields with -> in queries. section 11.6
Filter a schema-backed Connect record
A custom transformation obtains the original value with requireStruct, looks up or builds a reduced schema, creates updatedValue with that schema, and copies each retained field. Caching the revised schema avoids rebuilding the same record shape. section 5.6.3
Listing 5.14 Transforming with a schema
private R applyWithSchema(R connectRecord) {
final Struct value =
requireStruct(operatingValue(connectRecord), PURPOSE);
Schema updatedSchema = schemaUpdateCache.get(value.schema());
if(updatedSchema == null) {
updatedSchema = makeUpdatedSchema(value.schema());
schemaUpdateCache.put(value.schema(), updatedSchema);
}
final Struct updatedValue = new Struct(updatedSchema);
updatedValue.schema().fields()
.forEach(field -> updatedValue.put(field.name(),
value.get(field.name())));
return newRecord(connectRecord, updatedSchema, updatedValue);
}
Compare the schema-aware path with a Map-based schemaless transformation: the revised schema and revised Struct are produced together. section 5.6.3
Listing 11.40 Using the STRUCT datatype to represent nested data
CREATE STREAM school_event_stream (
event_id INT,
event STRUCT<type VARCHAR, #1
date VARCHAR,
student STRUCT<first_name VARCHAR, #2
last_name VARCHAR,
id BIGINT,
email VARCHAR
>,
class STRUCT<name VARCHAR,
room VARCHAR,
professor STRUCT<first_name VARCHAR,
last_name VARCHAR,
other_classes ARRAY<VARCHAR>
>
>
>
)
WITH (kafka_topic='school_events',
partitions=1,
value_format='JSON'
)
See nested JSON objects expressed as recursively nested STRUCT declarations for a ksqlDB stream. section 11.6
