Create import options based on event stream content
This function requires Streaming Data Framework for MATLAB® Production Server™.
obtains the event schema from an event stream and based on that schema, returns data import
options in the
opts = detectImportOptions(
opts. The schema includes the names and data types of variables in
the event body.
You can modify
opts later and use it with
to control how MATLAB imports stream events into MATLAB timetables.
analyzes the event structure array
opts = detectImportOptions(
e to determine the event schema. Use
this syntax only if the event stream that
stream connects to is
unavailable. You can either manually create this array, which can be complex, or use one
that was previously obtained from the stream, such as from a call to the
Detect Import Options from Event Stream
Assume that you have a Kafka® server running at the network address
that has a topic
Triangles. Each event has one data column,
"triangle", which contains a structure with three fields,
"z", containing the
integer side lengths of a triangle.
KafkaStream object connected to the
ks = kafkaStream("kafka.host.com",9092,"Triangles");
ImportOptions object from the Kafka stream object. The data type of the length of each side is
opts = detectImportOptions(ks)
opts = ImportOptions with properties: VariableNames: ["triangle/x" "triangle/y" "triangle/z"] VariableTypes: ["string" "string" "string"] KeyVariable: "key" SelectedVariableNames: ["triangle/x" "triangle/y" "triangle/z"]
To perform mathematical operations on the imported data, update the data type of variables
double. Because the side length variables are nested within
"triangle", use a forward slash (
"/") to specify the
path to these variables.
opts = setvartype(opts, ["triangle/x", "triangle/y", "triangle/z"], "double")
opts = ImportOptions with properties: VariableNames: ["triangle/x" "triangle/y" "triangle/z"] VariableTypes: ["double" "double" "double"] KeyVariable: "key" SelectedVariableNames: ["triangle/x" "triangle/y" "triangle/z"]
ImportOptions property of the
ks.ImportOptions = opts
ks = KafkaStream with properties: Topic: "Triangles" Group: "85c42e39-695d-467a-86f0-f0095792e7de" Order: EventTime Host: "kafka.host.com" Port: 9092 ConnectionTimeout: 30 RequestTimeout: 61 ImportOptions: "Import to MATLAB types" ExportOptions: "Source: function eventSchema" PublishSchema: "true" WindowSize: 50 KeyVariable: "key" KeyEncoding: "utf16" KeyType: "text" KeyByteOrder: "BigEndian" BodyEncoding: "utf8" BodyFormat: "JSON" ReadLimit: "Size" TimestampResolution: "Milliseconds"
When importing the triangles,
readtimetable converts the side lengths
tt = readtimetable(ks); tt(1,:).triangle
ans = struct with fields: x: 3 y: 4 z: 5
e — Event information
Event information, specified as a structure array. This array is in the format
returned by the
Each structure in the array has these fields.
key — Event key
string array | positive integer
Event key as stored in Kafka, returned as a string array or integer. The key identifies the event source.
value — Event value
Event value, specified as a byte array with a format and encoding determined by the
BodyEncoding properties of the stream object. The event value does not undergo schema processing and appears exactly as is stored in Kafka, for example, as a JSON string.
timestamp — Event timestamp or ingest timestamp
Timestamp of event occurrence or timestamp of event ingestion in Kafka, specified as a datetime scalar.
Introduced in R2022b