Aggregate

Aggregate transforms aggregate or disaggregate data.

aggregate.to.array

Transforms data by aggregating it into a JSON array.

Input

{"a":"b"}
{"c":"d"}
{"e":"f"}

Output

[{"a":"b"},{"c":"d"},{"e":"f"}]

Settings

Field	Type	Description	Required
batch.count	int	Maximum number of items to buffer before emitting a new array. Defaults to 1,000 items.	No
batch.size	int	Maximum size (in bytes) of items to buffer before emitting a new array. Defaults to 1MB.	No
batch.duration	string	Maximum duration to buffer items before emitting a new array. Defaults to 1m.	No
object.source_key	string	Retrieves a value from an object for transformation.	No
object.target_key	string	Places a value into an object after transformation.	No
object.batch_key	string	Retrieves a value from an object that is used to group batched data. No default, all data is batched into the same array.	No

Example

Config

sub.transform.aggregate.to.array()

Abbv. Config

sub.tf.agg.to.arr()

aggregate.from.array

Transforms data by disaggregating it from a JSON array. Values from the array become new messages.

Input

[{"a":"b"},{"c":"d"},{"e":"f"}]

Output

{"a":"b"}
{"c":"d"}
{"e":"f"}

Settings

Field	Type	Description	Required
object.source_key	string	Retrieves a value from an object for transformation.	No
object.target_key	string	Places a value into an object after transformation.	No

Example

Config

sub.transform.aggregate.from.array()

Abbv. Config

sub.tf.agg.from.arr()

aggregate.to.string

Transforms data by aggregating it to a string.

Input

{"a":"b"}
{"c":"d"}
{"e":"f"}

Output

{"a":"b"}\n{"c":"d"}\n{"e":"f"}

Settings

Field	Type	Description	Required
batch.count	int	Maximum number of items to buffer before emitting a new string. Defaults to 1,000 items.	No
batch.size	int	Maximum size (in bytes) of items to buffer before emitting a new string. Defaults to 1MB.	No
batch.duration	string	Maximum duration to buffer items before emitting a new string. Defaults to 5m.	No
object.source_key	string	Retrieves a value from an object for transformation.	No
object.batch_key	string	Retrieves a value from an object that is used to organize batched data. No default, all data is batched into the same string.	No
separator	string	Separator substring that is used to aggregate data.	Yes

Example

Config

sub.transform.aggregate.to.string(
  settings={separator: '\n'}
)

Abbv. Config

sub.tf.agg.to.str({separator: '\n'})

aggregate.from.string

Transforms data by disaggregating it from a string.

Input

{"a":"b"}\n{"c":"d"}\n{"e":"f"}

Output

{"a":"b"}
{"c":"d"}
{"e":"f"}

Settings

Field	Type	Description	Required
separator	string	Separator substring that is used to disaggregate data.	Yes

Example

Config

sub.transform.aggregate.from.string(
  settings={separator: '\n'}
)

Abbv. Config

sub.tf.agg.from.str({separator: '\n'})

Use Cases

Organizing Data by Key

The buffer.key setting can be used to dynamically organize data based on values inside of objects. For example, if the buffer.key is set to "key", then these results are produced by the aggregate.to.array transform:

Input

{"a":"b","key":"x"}
{"c":"d","key":"y"}
{"e":"f","key":"x"}

Output

[{"a":"b","key":"x"},{"e":"f","key":"x"}]
[{"c":"d","key":"y"}]