Working with non-SQL CDC readers

See JSONNode functions.

Use the PutUserData function in a CQ to add a field to the JSONNodeEvent USERDATA map. For examples of how to use USERDATA elements in TQL, see Modifying output using ColumnMap and the discussions of PartitionKey in Kafka Writer and S3 Writer.

The following example would add the sixth element (counting from zero) in the JSONNodeEvent userdata array to USERDATA as the field "city":

CREATE CQ AddUserData
INSERT INTO EnrichedStream
SELECT putUserData(x, 'city', data[5])
FROM MongoDBSourceStream x;

To remove an element from the USERDATA map, use the removeUserData function (you may specify multiple elements, separated by commas):

CREATE CQ RemoveUserData
INSERT INTO EnrichedStream 
SELECT removeUserData(x, 'city')
FROM MongoDBSourceStream x;

To remove all elements from the USERDATA map, use the clearUserData function:

CREATE CQ ClearUserData
INSERT INTO EnrichedStream 
SELECT clearUserData(x)
FROM MongoDBSourceStream x;

Using the application and data from MongoDBReader example application and output, the following CQ would convert the JSONNodeEvent data array values to a Striim type, which could then be consumed by any other Striim component.

CREATE CQ JSON2StriimType
INSERT INTO EmployeeStream
SELECT data.get("_id").toString() AS id,
  data.get("firstname").toString() AS firstname,
  data.get("lastname").toString() AS lastname,
  data.get("age").toString() AS age
FROM MongoDBStream;