json

README

JSON Processor

To use the JSON Processor use this data config:

data:
  processor:
    name: json

Params

Name	Supported Values	Description
time_format	A Golang time.Parse layout	Specifies the format of the time field for parsing
time_selector	A top-level field name	Specifies the field to use for time

Dataspace Config

In the current version, the JSON Processor supports a flat object structure. Fields are mapped into Observations using the measurements, categories and tags nodes. E.g.

dataspaces:
  - from: event
    name: stream
    data:
      processor:
        name: json
    measurements:
      - name: speed
      - name: height
      - name: target
    categories:
      - name: rating
        values:
          - 1
          - 2
          - 3
          - 4
          - 5
    tags:
      selectors:
        - tags
      values:
        - tagA
        - tagB

An example input payload is as follows:

[
  {
    "time": 1631499271,
    "rating": 4,
    "speed": 26,
    "height": 81,
    "target": 12,
    "tags": ["tagA", "tagB"]
  }
]

Datatypes

• The value of time must either be a Unix timestamp or a string conforming to RFC3339 i.e. 1985-04-12T23:20:50.52Z or a custom time format must be passed in the time_format param.

• Measurements must be JSON numbers.

• Categories must be strings.

• When using tags or _tags selectors, the tags field is parsed as an array of tag values. Any other field assumes a single string tag value.

Documentation

Index

• Constants
• type JsonProcessor
  • func NewJsonProcessor() *JsonProcessor
  • func (p *JsonProcessor) GetRecord() (arrow.Record, error)
  • func (p *JsonProcessor) Init(params map[string]string, identifiers map[string]string, ...) error
  • func (p *JsonProcessor) OnData(data []byte) ([]byte, error)

Constants

const (
	JsonProcessorName string = "json"
)

Types

type JsonProcessor

type JsonProcessor struct {
	// contains filtered or unexported fields
}

func NewJsonProcessor

func NewJsonProcessor() *JsonProcessor

func (*JsonProcessor) GetRecord

func (p *JsonProcessor) GetRecord() (arrow.Record, error)

func (*JsonProcessor) Init

func (p *JsonProcessor) Init(params map[string]string, identifiers map[string]string, measurements map[string]string, categories map[string]string, tags []string) error

func (*JsonProcessor) OnData

func (p *JsonProcessor) OnData(data []byte) ([]byte, error)