EdgeX Source

Kuiper provides built-in support for EdgeX source stream, which can subscribe the message from EdgeX message bus and feed into the Kuiper streaming process pipeline.

Stream definition for EdgeX

EdgeX already defines data types in value descriptors, so it's recommeded to use schema-less stream definition in EdgeX source as in below.

# cd $kuiper_base
# bin/cli CREATE STREAM demo'() with(format="json", datasource="demo" type="edgex")'

EdgeX source will try to get the data type of a field,

convert to releated data type if field of a type can be found in value descriptors service;
or keep original value if field of a type can not be found in value descriptors service;
or if failed to conver the value, then the value will be dropped, and a warning message print in the log;

The types defined in EdgeX value descriptors will be converted into related data types that supported in Kuiper.

Boolean

If Type value of ValueDescriptor is B, Bool or Boolean, then Kuiper tries to convert to boolean type. Following values will be converted into true.

"1", "t", "T", "true", "TRUE", "True"

Following will be converted into false.

"0", "f", "F", "false", "FALSE", "False"

Bigint

If Type value of ValueDescriptor is I, INT, INT8 , INT16, INT32, INT64,UINT , UINT8 , UINT16 , UINT32 , UINT64 then Kuiper tries to convert to Bigint type.

Float

If Type value of ValueDescriptor is F, FLOAT, FLOAT16 , FLOAT32, FLOAT64then Kuiper tries to convert to Float type.

String

If Type value of ValueDescriptor is S, String, then Kuiper tries to convert to String type.

Struct

If Type value of ValueDescriptor is J, Json, then Kuiper tries to convert to Struct type.

Global configurations

The configuration file of EdgeX source is at $kuiper/etc/sources/edgex.yaml. Below is the file format.

#Global Edgex configurations
default:
  protocol: tcp
  server: localhost
  port: 5570
  topic: events
  serviceServer: http://localhost:10080
#  optional:
#    ClientId: client1
#    Username: user1
#    Password: password

Use can specify the global EdgeX settings here. The configuration items specified in default section will be taken as default settings for all EdgeX source.

protocol

The protocol connect to EdgeX message bus, default value is tcp.

server

The server address of EdgeX message bus, default value is localhost.

port

The port of EdgeX message bus, default value is 5570.

topic

The topic name of EdgeX message bus, default value is events.

serviceServer

The base service address for getting value descriptors, the value of serviceServer will be concatenated to /api/v1/valuedescriptor to get all of value descriptors of EdgeX server.

Override the default settings

In some cases, maybe you want to consume message from multiple topics or event bus. Kuiper supports to specify another configuration, and use the CONF_KEY to specify the newly created key when you create a stream.

#Override the global configurations
demo1: #Conf_key
  protocol: tcp
  server: 10.211.55.6
  port: 5570
  topic: events

If you have a specific connection that need to overwrite the default settings, you can create a customized section. In the previous sample, we create a specific setting named with demo1. Then you can specify the configuration with option CONF_KEY when creating the stream definition (see stream specs for more info).

Sample

create stream demo1() WITH (FORMAT="JSON", type="edgex", CONF_KEY="demo1");

The configuration keys used for these specific settings are the same as in default settings, any values specified in specific settings will overwrite the values in default section.

edgex.md 4.0 KB History Raw