The Lenses SQL Snapshot engine accesses the data at the point in time the query is executed. This means, that for Apache Kafka, data added just after the query was initiated will not be processed.

Typical use cases are but are not limited to:

• Identifying a specific message.

• Identifying a particular transaction of payment that your system has processed

• Identifying all thermostats readings for a specific customer if you are working for an energy provider

• Counting transactions processed within a given time window.

Writing queries

Go to Workspace->Sql Studio, enter your query, and click run.

WHERE clause

WHERE clause allows you to define a set of logical predicates the data needs to match in order to be returned. Standard comparison operators are supported (>, >=, <, <=, =, and !=) as well as calling functions.

We are going to use the groceries table created earlier. Select all items purchased where the prices are greater or equal to 2.00:

SELECT
     name
     , price
FROM groceries
WHERE price >= 2.0

/* Output
Meridian Crunchy Peanut Butter              2.5
Activia fat free cherry yogurts             2
Green & Blacks Organic Chocolate Ice Cream  4.2
*/

Select all customers whose last name length equals to 5:

SELECT *
FROM customers
WHERE LEN(last_name) = 5

/* Output
key         value.first_name    value.last_name
mikejones       Mike                Jones
anasmith        Ana                 Smith
*/

Search all customers containing Ana in their first name:

SELECT *
FROM customers
WHERE first_name LIKE '%Ana%'

Keep in mind that text search is case-sensitive. To use case insensitive text search, you can write:

SELECT *
FROM customers
WHERE LOWERCASE(first_name) LIKE '%ana%';

-- And here is the negated version
SELECT *
FROM customers
WHERE LOWERCASE(first_name) NOT LIKE '%ana%';

Missing values

Sometimes data can contain explicit NULL values, or it can omit fields entirely. Using IS [ NOT ] NULL, or EXISTS functions allows you to check for these situations.

Exists is a keyword in Lenses SQL grammar so it needs to be escaped, the escape character is `````.

Lenses supports JSON. JSON does not enforce a schema allowing you to insert null values.

Create the following table named customers_json:

CREATE TABLE customers_json (
    _key STRING
    , first_name STRING
    , last_name STRING
    , middle_name STRING
) FORMAT(string, json);


INSERT INTO customers_json(_key, first_name, last_name, middle_name) VALUES("mikejones", "Mike", "Jones", "Albert");
INSERT INTO customers_json(_key, first_name, last_name) VALUES("anasmith", "Ana", "Smith");
INSERT INTO customers_json(_key, first_name, last_name) VALUES("shannonelliott", "Shannon","Elliott");

Query this table for all its entries:

SELECT * 
FROM customers_json

/* The output
key             value.first_name   value.middle_name   value.last_name
mikejones           Mike                Albert          Jones
anasmith            Ana                                 Smith
shannonelliott      Shannon                             Elliott
*/

The middle_name is only present on the mikejones record.

Write a query which filters out records where middle_name is not present:

SELECT *
FROM customers_json
WHERE `EXISTS`(middle_name)

/* The output
 key             value.first_name   value.middle_name   value.last_name
mikejones            Mike            Albert                Jones
*/

This can also be written as:

SELECT *
FROM customers_json
WHERE middle_name IS NULL

When a field is actually NULL or is missing, checking like in the above query has the same outcome.

Multiple WHERE conditions

You can use AND/OR to specify complex conditions for filtering your data.

To filter the purchased items where more than one item has been bought for a given product, and the unit price is greater than 2:

SELECT *
FROM groceries
WHERE quantity > 1 
    AND price > 2

Now try changing the AND logical operand to OR and see the differences in output.

HAVING clause

To filter the entries returned from a grouping query. As with the WHERE statement, you can use HAVING syntax to achieve the same result when it comes to grouped queries.

SELECT
    COUNT(*) AS count
    , country
FROM customer
GROUP BY country
HAVING count > 1

Read a table partition only

To select data from a specific partition access the metadata of the topic.

In the following example, a table is created with three partitions and the message key is hashed and then the remainder HashValue % partitions will be the table partition the record is sent to.

-- Run
CREATE TABLE customers_partitioned (
    _key STRING
    , first_name STRING
    , last_name STRING
) 
FORMAT(string, Avro)
properties(partitions = 3);

INSERT INTO customers_partitioned(
    _key
    , first_name
    , last_name)
VALUES
("mikejones", "Mike", "Jones"),
("anasmith", "Ana", "Smith"),
("shannonelliott", "Shannon","Elliott"),
("tomwood", "Tom","Wood"),
("adelewatson", "Adele","Watson"),
("mariasanchez", "Maria", "Sanchez");

Next, run the following query:

SELECT *
FROM customers_partitioned

/* The output
offset  partition   timestamp       key         value.first_name    value.last_name
0       0           1540830780401   mikejones       Mike                Jones
1       0           1540830780441   anasmith        Ana                 Smith
2       0           1540830780464   shannonelliott  Shannon             Elliott
0       2           1540831270170   mariasanchez    Maria               Sanchez
0       1           1540830984698   tomwood         Tom                 Wood
1       1           1540831183308   adelewatson     Adele               Watson
*/

As you can see from the results (your timestamps will be different) the records span over the three partitions. Now query specific partitions:

-- selects only records from partition = 0
SELECT *
FROM customers_partitioned
WHERE _meta.partition = 0;

-- selects only records from partition  0 and 2
SELECT *
FROM customers_partitioned
WHERE _meta.partition = 0
   OR _meta.partition = 2;

Query by partitions

Kafka reads are non-deterministic over multiple partitions. The Snapshot engine may reach its max.size before it finds your record in one run, next time it might.

SELECT *
FROM topicA
WHERE transaction_id=123
   AND _meta.partition = 1

If we specify in our query that we are only interested in partition 1, and for the sake of example the above Kafka topic has 50 x partitions. Then Lenses will automatically push this predicate down, meaning that we will only need to scan 1GB instead of 50GB of data.

Query by offsets

SELECT *
FROM topicA
WHERE transaction_id=123
  AND _meta.offset > 100
  AND _meta.offset < 100100
  AND _meta.partition = 1

If we specify the offset range and the partition, we would only need to scan the specific range of 100K messages resulting in scanning 5MB.

Query by timestamp

SELECT *
FROM topicA
WHERE transaction_id=123
  AND _meta.timestamp > NOW() - "1H"

The above will query only the data added to the topic up to 1 hour ago. Thus we would query just 10MB.Time-traveling

SELECT *
FROM position_reports
WHERE
   _meta.timestamp > "2020-04-01" AND
   _meta.timestamp < "2020-04-02"

The above will query only the data that have been added to the Kafka topic on a specific day. If we are storing 1,000 days of data, we would query just 50MB.

Does Apache Kafka have indexing?

No. Apache Kafka does not have the full indexing capabilities in the payload (indexes typically come at a high cost even on an RDBMS / DB or a system like Elastic Search), however, Kafka indexes the metadata.

The only filters Kafka supports are topic, partition and offsets or timestamps.

When querying Kafka topic data with SQL such as

SELECT *
FROM topicA
WHERE transaction_id=123

a full scan will be executed, and the query processes the entire data on that topic to identify all records that match the transaction id.

If the Kafka topic contains a billion 50KB messages - that would require querying 50 GB of data. Depending on your network capabilities, brokers’ performance, any quotas on your account, and other parameters, fetching 50 GB of data could take some time! Even more, if the data is compressed. In the last case, the client has to decompress it before parsing the raw bytes to translate into a structure to which the query can be applied.

Understanding bad records

When Lenses can’t read (deserialize) your topic’s messages, it classifies them as “bad records”. This happens for one of the following reasons:

• Kafka records are corrupted. On an AVRO topic, a rogue producer might have published a different format

• Lenses topic settings do not match the payload data. Maybe a topic was incorrectly given AVRO format when it’s JSON or vice versa

• If AVRO payload is involved, maybe the Schema Registry is down or not accessible from the machine running Lenses

By default, Lenses skips them and displays the records’ metadata in the Bad Records tab. If you want to force stop the query in such case use:

SET skip.bad.records=false;
SELECT * FROM topicA LIMIT 100

Control execution

Querying a table can take a long time if it contains a lot of records. The underlying Kafka topic has to be read, the filter conditions applied, and the projections made.

Additionally, the SELECT statement could end up bringing a large amount of data to the client. To be able to constrain the resources involved, Lenses allows for context customization, which ends up driving the execution, thus giving control to the user. Here is the list of context parameters to overwrite:

All the above values can be given a default value via the configuration file. Using lenses.sql.settings as prefix the format.timestamp can be set like this:

lenses.sql.settings.format.timestamp=true

Query tuning

Lenses SQL uses Kafka Consumer to read the data. This means that an advanced user with knowledge of Kafka could tweak the consumer properties to achieve better throughput. This would occur on very rare occasions. The query context can receive Kafka consumer settings. For example, the max.poll.records consumer can be set as:

SET max.poll.records= 100000;

SELECT *
FROM payments
LIMIT 1000000

Example

The fact is that streaming SQL is operating on unbounded streams of events: a query would normally be a never-ending query. In order to bring query termination semantics into Apache Kafka we introduced 4 controls:

• LIMIT = 10000 - Force the query to terminate when 10,000 records are matched.

• max.bytes = 20000000 - Force the query to terminate once 20 MBytes have been retrieved.

• max.time = 60000 - Force the query to terminate after 60 seconds.

• max.zero.polls = 8 - Force the query to terminate after 8 consecutive polls are empty, indicating we have exhausted a topic.

Thus, when retrieving data, you can set a limit of 1GB to the maximum number of bytes retrieved and a maximum query time of one hour like this:

SET max.bytes = 1000000000;
SET max.time = 60000000;

SELECT * 
FROM topicA 
WHERE customer.id = "XXX";

Snapshot queries on streaming data provide answers to a direct question, e.g. The current balance is $10. The query is active, the data is passive.

What is a message?

A single entry in a Kafka topic is called a message.

The engine considers a message to have four distinct components key, value, headers and metadata.

Facets

Currently, the Snapshot Engine supports four different facets _key, _value, _headers and _metadata; These strings can be used to reference properties of each of the aforementioned message components and build a query that way.

By default, unqualified properties are assumed to belong to the _value facet:

SELECT 
  property
FROM source_topic;

In order to reference a different facet, a facet qualifier can be added:

SELECT 
  _value.valueField,
  _key.keyField,
  _meta.metaField,
  _headers.headerField
FROM source_topic;

When more than one sources/topics are specified in a query (like it happens when two topics are joined) a table reference can be added to the selection to fix the ambiguity:

SELECT 
  users._value.field
FROM users JOIN purchases

the same can be done for any of the other facets (_key,_meta,_headers).

Note Using a wildcard selection statement SELECT * provides only the value component of a message.

Selecting nested fields

Messages can contain nested elements and embedded arrays. The . operator is used to refer to children, and the [] operator is used for referring to an element in an array.

You can use a combination of these two operators to access data of any depth.

SELECT 
    dependencies[0].first_name AS childName
FROM policy_holder
WHERE policyId='100001'

You explicitly reference the key, value and metadata.

For the key use _key, for the value use _value, and for metadata use _meta. When there is no prefix, the engine will resolve the field(s) as being part of the message value. For example, the following two queries are identical:

SELECT 
    amount
FROM payments;

SELECT 
    _value.amount
FROM payments;

Primitive types

When the key or a value content is a primitive data type use the prefix only to address them.

For example, if messages contain a device identifier as the key and the temperature as the value, SQL code would be:

SELECT 
    _key AS deviceId
  , _value AS temperature
FROM iot_data

Accessing metadata

Use the _meta keyword to address the metadata. For example:

SELECT 
    _meta.timestamp AS timestamp
    , _meta.offset AS index
FROM iot_data

Projections and nested aliases

When projecting a field into a target record, Lenses allows complex structures to be built. This can be done by using a nested alias like below:

SELECT 
    amount as user.amount
    userId as user.id
FROM payments;

The result would be a struct with the following shape:

{
  "user": {
    "amount" : 10.19,
    "id": 10
  }  
}

Alias clashes (repeated fields)

When two alias names clash, the snapshot engine does not “override” that field. Lenses will instead generate a new name by appending a unique integer. This means that a query like the following:

SELECT 
    amount as result.amount,
    amount + 5 as result.amount
FROM payments;

will generate a structure like the following:

{
  "result": {
    "amount" : 10, 
    "amount0": 15
  }  
}

Nested queries

The tabled query allows you to nest queries. Let us take the query in the previous section and say we are only interested in those entries where there exist more than 1 customer per country.

SELECT *
FROM (
    SELECT 
        COUNT(*) AS count
        , country
    FROM customer
    GROUP BY country
    )
WHERE count > 1

Run the query, and you will only see those entries for which there is more than one person registered per country.

Functions

Functions can be used directly.

For example, the ROUND function allows you to round numeric functions:

SELECT
    name 
    , ROUND(quantity * price) AS rounded_total
FROM groceries

/*The output:
Fairtrade Bananas                                   2
Meridian Crunchy Peanut Butter                      3
Green & Black's organic 85% dark chocolate bar      4
Activia fat free cherry yogurts                     6
Green & Blacks Organic Chocolate Ice Cream          8
*/

Limit the output

To limit the output of the query you can use two approaches:

• use the LIMIT clause

• set the max size of the data to be returned

-- limit to 1 record
SELECT *
FROM groceries
LIMIT 1

-- set the data size returned to be 1 megabyte.
SET max.size="1m";

-- on the small dataset we have here 1 MB will accommodate all records added and more
SELECT *
FROM groceries

Set a time limit for a query

To restrict the time to run the query, use SET max.query.time:

SET  max.query.time = '1h';

SELECT ...
    FROM table
    LIMIT 1000

Sampling data

To sample data and discard the first rows:

SELECT *
FROM groceries
LIMIT 1,2

This statement instructs Lenses to skip the first record matched and then sample the next two.

Lenses SQL allows you to utilize the ANSI SQL command to store new records into a table.

Single or multi-record inserts are supported:

INSERT INTO $Table(column1[, column2, column3])
VALUES(value1[,value2, value3])

INSERT INTO $Table(column1[, column2, column3])
VALUES
(value1[,value2, value3]),
(value4[,value5, value6])

• $Table - The name of the table to insert the data into

• Columns - The target columns to populate with data. Adding a record does not require you to fill all the available columns. In the case of Avro stored Key, Value pairs, the user needs to make sure that a value is specified for all the required Avro fields.

• VALUES - The set of value to insert. It has to match the list of columns provided, including their data types. You can use simple constants or more complex expressions as values, like 1 + 1 or NOW().

Example:

INSERT INTO customer (
    _key, id
    , address.line
    , address.city
    , address.postcode
    , email)
VALUES
('maria.wood','maria.wood', '698 E. Bedford Lane','Los Angeles', 90044, 'maria.wood@lenses.io'),
('david.green', 'david.green', '4309 S Morgan St', 'Chicago', 60609, 'david.green@lenses.io');

Inserting data from a SELECT

Records can be inserted from the result of SELECT statement.

The syntax is:

INSERT INTO $TABLE1
SELECT */column1[,column2, ...]
FROM $Table2
[WHERE $condition]
[LIMIT N]

For example, to copy all the records from the customer table into customer_avro one:

INSERT INTO customer_avro
SELECT *
FROM customer

Insert complex key

There are scenarios where a record key is a complex type. Regardless of the storage format, JSON or Avro, the SQL engine allows the insertion of such entries:

-- creates a smart_devices table where the key is a type with one field {deviceId:100}
CREATE TABLE smart_devices(
                              _key.deviceId INT
    , room INT
    , temperature double)
    FORMAT(avro, avro);

INSERT INTO smart_devices(
                           _key.deviceId
                         , room
                         , temperature)
VALUES(11223, 99, 22.1);

SELECT *
FROM smart_devices;

Deleting data in Kafka

There are two ways to delete data:

• If the topic is not compacted, then DELETE expects an offset to delete records up to.

-- Delete records across all partitions up to the offset 10
DELETE FROM customer 
WHERE _meta.offset <= 10;

-- Delete records from a specific partition 
DELETE FROM customer 
WHERE _meta.offset <= 10 AND _meta.partition = 2

• If the topic is compacted, then DELETE expects the record Key to be provided. For a compacted topic a delete translates to inserting a record with the existing Key, but the Value is null. For the customer_avro topic (which has the compacted flag on), a delete operation for a specific customer identifier would look like this:

DELETE FROM customer_avro
WHERE _key = 'andy.perez'

Deleting is an insert operation. Until the compaction takes place, there will be at least one record with the Key used earlier. The latest (or last) record will have the Value set to null.

Truncating a table

To remove all records from a table:

TRUNCATE TABLE $Table

where the $Table is the table name to delete all records from. This operation is only supported on non-compacted topics, which is a Kafka design restriction. To remove the data from a compacted topic, you have two options: either dropping and recreating the topic or inserting null Value records for each unique Key on the topic.

After rebuilding the customer table to be non-compacted, perform the truncate:

TRUNCATE TABLE customer;
/* SELECT count(*) FROM customer returns 0 after the previous statement */

Lenses supports the typical SQL commands supported by a relational database:

• CREATE

• DROP

• TRUNCATE

• DELETE

• SHOW TABLES

• DESCRIBE TABLE

• DESCRIBE FORMATTED

CREATE TABLE

CREATE TABLE 
    table_name(
        $field $fieldType [, $field $fieldType,...]
    )
FORMAT ($keyStorageFormat, $valueStorageFormat)
[PROPERTIES(
        partitions= *, 
        replication=$replication, 
        compacted=true/false)
];

The CREATE statement has the following parts:

• CREATE TABLE - Instructs the construction of a table

• $Table - The actual name given to the table.

• Schema - Constructed as a list of (field, type) tuple, it describes the data each record in the table contains

• FORMAT - Defines the storage format. Since it is an Apache Kafka topic, both the Key and the Value formats are required. Valid values are STRING, INT, LONG, JSON, AVRO.

• PROPERTIES - Specifies the number of partitions the final Kafka topic should have, the replication factor in order to ensure high availability (it cannot be a number higher than the current Kafka Brokers number) and if the topic should be compacted.

A Kafka topic which is compacted is a special type of topic with a finer-grained retention mechanism that retains the last update record for each key.

A compacted topic (once the compaction has been completed) contains a full snapshot of the final record values for every record key and not just the recently changed keys. They are useful for in-memory services, persistent data stores, reloading caches, etc.

For more details on the subject, you should look at Kafka Documentation.

Example:

CREATE TABLE customer (
        id STRING 
        , address.line STRING 
        , address.city STRING, 
        , address.postcode INT 
        , email STRING
    )
FORMAT (string, json)
PROPERTIES (
    partitions=1, 
    compacted=true
);

Best practices dictate to use Avro as a storage format over other formats. In this case, the key can still be stored as STRING but the value can be Avro.

CREATE TABLE customer_avro (
        id STRING
        , address.line STRING
        , address.city STRING 
        , address.postcode int 
        , email string
    )
FORMAT (string, avro)
PROPERTIES (
    partitions=1, 
    compacted=true
);

SHOW TABLES

To list all tables:

SHOW TABLES;

DESCRIBE TABLE

To examine the schema an metadata for a topic:

DESCRIBE TABLE $tableName

The $tableName should contain the name of the table to describe.

Given the two tables created earlier, a user can run the following SQL to get the information on each table:

DESCRIBE TABLE customer_avro

the following information will be displayed:

/* Output:
# Column Name                           # Data Type
_key                                    String
_value.address.postcode                 Int
_value.address.city                     String
_value.address.line                     String
_value.email                            String
_value.id                               String

# Config Key                            # Config Value
cleanup.policy                          compact
compression.type                        producer
delete.retention.ms                     86400000
file.delete.delay.ms                    60000
flush.messages                          9223372036854775807
flush.ms                                9223372036854775807
index.interval.bytes                    4096
max.message.bytes                       1000012
message.format.version                  1.1-IV0
message.timestamp.difference.max.ms     9223372036854775807
message.timestamp.type                  CreateTime
min.cleanable.dirty.ratio               0.5
min.compaction.lag.ms                   0
min.insync.replicas                     1
preallocate                             false
retention.bytes                         2147483648
retention.ms                            604800000
segment.bytes                           1073741824
segment.index.bytes                     10485760
segment.jitter.ms                       0
segment.ms                              604800000
unclean.leader.election.enable          false
*/

DROP TABLE

To drop a table:

DROP TABLE $Table;

System virtual tables (__tables, __fields)

Lenses provides a set of virtual tables that contain information about all the fields in all the tables.

Using the virtual table, you can quickly search for a table name but also see the table type.

The __table has a table_name column containing the table name, and a table_type column describing the table type (system, user, etc).

SELECT * FROM __tables;

SELECT *
FROM __tables
WHERE table_name LIKE '%customer%';

To see all the tables fields select from the _fields virtual table.

SELECT * 
FROM __fields;

SELECT * 
FROM __fields
WHERE table_name LIKE '%customer%'

Record metadata

Each Kafka message contains information related to partition, offset, timestamp, and topic. Additionally, the engine adds the key and value raw byte size.

Create a topic and insert a few entries.

CREATE TABLE tutorial(
    _key STRING
  , name STRING
  , difficulty INT
) 
FORMAT (Avro, Avro);

INSERT INTO tutorial(_key, name, difficulty)
VALUES
("1", "Learn Lenses SQL", 3),
("2", "Learn Quantum Physics", 10),
("3", "Learn French", 5);

Now we can query for specific metadata related to the records.

To query for metadata such as the underlying Kafka topic offset, partition and timestamp prefix your desired fields with _meta.

Run the following query to see each tutorial name along with its metadata information:

SELECT name
    , _meta.offset
    , _meta.timestamp
    , _meta.partition
    , _meta.__keysize
    , _meta.__valsize
FROM tutorial

/* The output is (timestamp will be different)
Learn Lenses SQL         0    1540575169198    0    7   23
Learn Quantum Physics    1    1540575169226    0    7   28
Learn French             2    1540575169247    0    7   19
*/

Count the records

Using the COUNT aggregate function you can count the records in a table. Run the following SQL to see how many records we have on the customers_partitioned:

SELECT 
    COUNT(*) AS total
FROM customers_partitioned

Use SUM to aggregate your amounts

Using the SUM function you can sum records in a table.

SELECT 
    SUM(quantity * price) AS amount
FROM groceries

Group data with GROUP BY

To group data use the GROUP BY clause:

CREATE TABLE customer (
    firstName STRING
    , lastName STRING
    , city STRING
    , country STRING
    , phone STRING) 
FORMAT(string, avro) 
properties(compacted=true);

INSERT INTO customer (
    _key
    ,firstName
    , lastName
    , city
    , country
    , phone)
VALUES
('1','Craig', 'Smith', 'New York', 'USA', '1-01-993 2800'),
('2','William', 'Maugham','Toronto','Canada','+1-613-555-0110'),
('3','William', 'Anthony','Leeds','UK', '+44 1632 960427'),
('4','S.B.','Swaminathan','Bangalore','India','+91 7513200000'),
('5','Thomas','Morgan','Arnsberg','Germany','+49-89-636-48018'),
('6','Thomas', 'Merton','New York','USA', '+1-617-555-0147'),
('7','Piers','Gibson','London','UK', '+44 1632 960269'),
('8','Nikolai','Dewey','Atlanta','USA','+1-404-555-0178'),
('9','Marquis', 'de Ellis','Rio De Janeiro','Brazil','+55 21 5555 5555'),
('10','Joseph', 'Milton','Houston','USA','+1-202-555-0153'),
('11','John','Betjeman Hunter','Sydney','Australia','+61 1900 654 321'),
('12','Evan', 'Hayek','Vancouver','Canada','+1-613-555-0130'),
('13','E.','Howard','Adelaide','Australia','+61 491 570 157'),
('14','C. J.', 'Wilde','London','UK','+44 1632 960111'),
('15','Butler', 'Andre','Florida','USA','+1-202-555-0107');

Let’s see how many customers there are from each country. Here is the code which computes that:

SELECT 
    COUNT(*) AS count
    , country
FROM customer
GROUP BY country

/* Output
2    Canada
3    UK
2    Australia
1    India
1    Brazil
5    USA
1    Germany
*/

SQL Processors continuously process data in Kafka topics. Under the hood, the Kafka Streams APIs are combined with the internal Lenses application deployment framework. A user gets a seamless experience for creating scalable processing components to filter, aggregate, join, and transform Kafka topics, which can scale natively to Kubernetes.

SQL Processors are continuous queries, they process data as it arrives and react to events, e.g. payment made, $5 to Coffeshop. The data is active, the query is passive.

SQL Processors offer:

• A no-code, stand-alone application executing a given Lenses SQL query on current and future data

• Query graph visualisation

• Fully integrated experience within Lenses

• ACLs and Monitoring functionality out-of-the-box

• Ability to scale up and down workflows via Kubernetes

Creating a SQL Processor

To create SQL Processors go to Workspace->Apps->New App->SQL Processor.

1. Enter a name

2. Enter your SQL statement, the editor will help you with IntelliSense

3. Optionally specify a description, tag and ProcessorID (consumer group ID)

4. Select the deployment target, Kubernetes this will be a Kubernetes cluster and namespace.

If the configuration is valid the SQL Processor will be created. You can then click Start to deploy. Processors are not started automatically, click Start in the Actions menu to start the Processor.

Starting a SQL Processor

To start a Processor, select Start from the Action menu.

Viewing SQL Processors

Select the SQL processor. Lenses will show an overview of the health of the processor.

Selecting a Runner will display further information about each individual runner.

The metadata in the Summary tab also contains the consumer group ID (Processor ID).

The SQL tab shows the SQL Streaming statement the processor is running. A visual representation is shown in the Data Flow tab.

The Configuration tab shows the low-level settings.

Scaling a SQL Processor

To scale a processor, Select Scale from the Actions menu, and enter the desired amount of runners. In Kubernetes mode, the runners are pods.

Stopping a SQL Processor

Select Stop in the Actions Menu.

Snippets

Lenses provides helpful snippets for common scenarios. Select the snippet from the Help section.

Processor ID vs Application ID

ProcessorID is the public unique identifier for an SQL processor. It is customizable, meaning that you, as a user, have control over it and can set this identifier to any arbitrary string.

Restrictions on custom ProcessorIDs:

• They have to be unique across all processors

• Match the following regex: ^[a-zA-Z0-9\-\._]+:

  • Only letters, capital letters, numbers and -, _ & - are allowed.

  • It has to start with a letter or a number.

  • It cannot be empty.

One important aspect of the ProcessorID is that it is used as the Kafka consumer group identifier. That means that, in practice, this is the value that allows an SQL processor to build its consumer group and coordinate record ingestion from Kafka between all Processor replicas. Consequently, if the ProcessorID of a given SQL processor is changed, that processor will restart consuming messages from the beginning of the existing records in the topic.

The ApplicationID is the Lenses unique identifier, is automatically created by Lenses, and cannot be customized.

This is unique among all applications types; it does not matter if it’s an SQL processor or a different (new) sort of future application.

Lenses uses the ApplicationID to manage applications. This means that, when Starting, Stopping or Scaling an application, Lenses will use this attribute to pick the right instance.

Lenses supports the typical SQL commands supported by a relational database:

• CREATE

• DROP

• TRUNCATE

• DELETE

• SHOW VIEWS

A view is a virtual table, generated dynamically based on the results of a SELECT statement.

A view looks and acts just like a real table, but is always created on the fly as required, so it is always up to date.

A synonym is an alias for a table. This is useful if you have a topic with a long, unwieldy name like customer_reports_emea_april_2018_to_march_2018 and you want to access this as customer_reports.

CREATE VIEW

To create a view:

CREATE VIEW <viewname> AS <query>

Where viewname is the name of the virtual table that is used to access the records in the view, and the query is a standard SELECT statement.

Then we can query the view:

SELECT *
FROM customer_emails

A view acts as a virtual table. This means that a view can be filtered even more or that a projection can be applied to a view:

SELECT *
FROM customer_emails
WHERE name LIKE 'sam%'

DROP VIEW

To delete a view:

DROP VIEW <viewname>

If you wish to modify an existing view, use the syntax above to delete it, and then create a new view with the same name.

SHOW VIEW

To see a definition of a view. You can use the following syntax:

SHOW VIEW <viewname>

CREATE SYNONYM

To create a synonym:

CREATE SYNONYM <name> FOR <table>

DROP SYNONYM

To delete a synonym:

DROP SYNONYM <name>

If you wish to modify an existing synonym, use the syntax above to delete it, and then create a new synonym with the same name.

Common use cases

Three common reasons for using a view are:

• creating a projection from a table with a large number of fields

• representing joins as a single table

• and creating a preset filter

We will cover each scenario with an example.

Projection subset

If we have a table called customers which contains full customer records - name, email, age, registration date, country, password, and many others – and we find ourselves repeatedly querying it for just name and email.

A view could be created that returns just the name and email as a projection.

CREATE VIEW customer_emails AS
SELECT 
    name
    , email
FROM customers

There is no reason to specify the projection each time.

The benefit is more significant when we want to select a higher number of fields - say a topic with 50 fields, and we want to select only 15.

Representing joins

The statement that is used to generate the view can consist of one or more tables. One use case of views is to represent joined tables as if they were a single table. This avoids the need for writing a complex join query each time.

CREATE VIEW orders AS
    SELECT 
        c.customer_id
        , c.customer_name
        , c.customer_email
        , o.order_id
        , o.order_date
        , a.address
        , a.postcode
    FROM customers c JOIN orders o ON c.customer_id = o.customer_id
        JOIN addresses a ON o.delivery_address_id = a.address_id

Then we can select from this join like this:

SELECT *
FROM orders

Preset filter

Finally, another use case is to define a filter that is commonly used. If a topic contains transactions, and we often found ourselves searching for transactions from the UK. We could run this query each time:

SELECT *
FROM transactions
WHERE country = "UK"

Alternatively, we can set up a view with this filter pre-applied:

CREATE VIEW transactions_uk AS
SELECT *
FROM transactions
WHERE country = "UK"

Then use a SELECT query:

SELECT *
FROM transactions_uk

Termination Control

SELECT * FROM topicA WHERE _key.deviceId=123 LIMIT 10

Adding a LIMIT 10 in the SQL query will result in the SQL terminating early, as soon as 10 x messages have been discovered. It’s not a perfect solution as we might never find 10 x messages, and thus perform a full scan.

You can also set a maximum query or idle time:

SET max.query.time = 30s;

or max idle time, the idea is that there is no reason to keep polling if we have exhausted the entire topic:

SET max.idle.time = 5s;

or a maximum amount of data to read from Kafka. This controls how much data to read from Kafka NOT the required memory.

SET max.size = 1M;

Recent queries

Recent queries are displayed, but only for the current session, they are not currently retained.

Click on the play button to run a previous query. If a query is already running, you will be asked if you want to stop it first.

View All queries

SHOW ALL QUERIES

View Running queries

You can see all running queries by Lenses users using SQL:

SHOW QUERIES

Kill Running queries

You can force stop a query by another user using SQL:

KILL QUERY <id>

Create an array field

You can create array fields using the ..[] syntax:

CREATE TABLE table(
                      _key INT,
    , fialdA INT[]            -- Simple array
    , fieldB.fieldC STRING[]) -- Array in a subfield
    , fieldD INT[][]          -- Nested array
FORMAT(avro, avro);

Select an array field

Tables can store data containing arrays. Here is a SQL statement for querying an array item:

SELECT
    fieldA[1]
     , fieldB.fieldC[2].x
FROM table
WHERE fieldA[1] LIKE '%Lenses%'

When working with arrays is good to check the array bounds. See the SIZEOF function in the list of supported functions.

Query for array size

Sometimes you want to find out how many items are in your array. To do so you can run:

SELECT
    SIZEOF(arrayFieldA)
FROM table

Joins

Lenses allows you to combine records from two tables. A query can contain zero, one or multiple JOIN operations.

Create an orders table and insert some data into it:

CREATE TABLE orders(
    _key INT
    , orderDate STRING
    , customerId STRING
    , amount DOUBLE
) 
FORMAT(int, avro);

INSERT INTO orders (
    _key
    , orderDate
    , customerId
    , amount
)
VALUES
(1, '2018-10-01', '1', 200.50),
(2, '2018-10-11', '1', 813.00),
(3, '2018-10-11', '3', 625.20),
(4, '2018-10-11', '14', 730.00),
(5, '2018-10-11', '10', 440.00),
(6, '2018-10-11', '9', 444.80);

With these tables in place, join them to get more information about an order by combining it with the customer information found in the customer table:

SELECT 
    o._key AS orderNumber
    , o.amount AS totalAmount
    , c.firstName
    , c.lastName
    , c.city
    , c.country
FROM orders o INNER JOIN customer c
    ON o.customerId = c._key

/*
city        orderNumber     country     totalAmount     lastName    firstName
New York        1               USA         200.5       Smith       Craig
New York        2               USA         813         Smith       Craig
Leeds           3               UK          625.2       Anthony     William
Rio De Janeiro  6               Brazil      444.8       de Ellis    Marquis
Houston         5               USA         440         Milton      Joseph
London          4               UK          730         Wilde       C. J.
*/

Lateral Joins

With lateral joins, Lenses allows you to combine records from a table with the elements of an array expression.

We are going to see in more detail what lateral joins are with an example.

Create a batched_readings table and insert some data into it:

CREATE TABLE batched_readings(
  meter_id int
  , readings int[]
)
FORMAT(int, AVRO);

INSERT INTO batched_readings(
    meter_id
    , readings
) VALUES
(1, [100, 80, 95, 91]),
(2, [87, 93, 100]),
(1, [88, 89, 92, 94]),
(2, [81])

You now can use a LATERAL join to inspect, extract and filter the single elements of the readings array, as if they were a normal field:

SELECT
    meter_id
    , reading
 FROM
    batched_readings
    LATERAL readings AS reading
WHERE 
    reading > 90

Running that query we will get the values:

You can use multiple LATERAL joins, one inside the other, if you want to extract elements from a nested array:

CREATE TABLE batched_readings_nested(
  sensor_id int
  , nested_readings int[][]
)
FORMAT(int, AVRO);

INSERT INTO batched_readings_nested(
    sensor_id
    , nested_readings
) VALUES
(1, [[100, 101], [103]]),
(2, [[80, 81], [82, 83, 82]]),
(1, [[100], [103, 102], [104]])

Running the following query we will obtain the same records of the previous example:

SELECT
    meter_id
    , reading
 FROM
    batched_readings
    LATERAL nested_readings AS readings
    LATERAL readings as reading
WHERE 
    reading > 90

Joins allow rows from different sources to be combined.

Lenses allows two sources of data to be combined based either on the equality of their _key facet or using a user-provided expression.

A query using joins looks like a regular query apart from the definition of its source and in some cases the need to specify a window expression:

SELECT (STREAM|TABLE)
  <projection>
FROM
  <sourceA> [LEFT|RIGHT|INNER|OUTER] JOIN
  <sourceB> [ON <joinExpression>]
  [WITHIN <duration>]
WHERE
  <filterExpression>;

• projection: the projection of a join expression differs very little from a regular projection. The only important consideration is that since data is selected from two sources, some fields may be common to both. The syntax table.field is recommended to avoid this type of problem.

• sourceA/sourceB : the two sources of data to combine.

• window: only used if two streams are joined. Specifies the interval of time to search matching results.

• joinExpression: a boolean expression that specifies how the combination of the two sources is calculated.

• filterExpression: a filter expression specifying which records should be filtered.

Join types

When two sources of data are combined it is possible to control which records to keep when a match is not found:

Disclaimer: The following examples do not take into consideration windowing and/or table materialization concerns.

Customers

Orders

Inner join

WITH customersTable AS (SELECT TABLE * FROM customers);
WITH ordersStream AS (SELECT STREAM * FROM orders);

INSERT INTO result
SELECT STREAM
    customersTable.name
   , ordersStream.item
 FROM
    ordersStream JOIN customersTable
        ON customersTable.id = ordersStream.customer_id;

This join type will only emit records where a match has occurred.

(Notice there’s no item with customer.id = 2 nor is there a customer with id = 3 so these two rows are not present in the result).

Left join

WITH customersTable AS (SELECT TABLE * FROM customers);
WITH ordersStream AS (SELECT STREAM * FROM orders);

INSERT INTO result
SELECT STREAM
    customersTable.name
    , ordersStream.item
 FROM
    ordersStream LEFT JOIN customersTable
        ON customersTable.id = ordersStream.customer_id;

This join type selects all the records from the left side of the join regardless of a match:

(Notice all the rows from orders are present but since no customer.id = 3 no name can be set.)

Right join

WITH customersTable AS (SELECT TABLE * FROM customers);
WITH ordersStream AS (SELECT STREAM * FROM orders);

INSERT INTO result
SELECT TABLE
    customersTable.name
    , ordersStream.item
 FROM
    customersTable RIGHT JOIN ordersStream
        ON customersTable.id = ordersStream.customer_id;

A right join can be seen as a mirror of a LEFT JOIN. It selects all the records from the right side of the join regardless of a match:

Outer Join

WITH customersStream AS (SELECT STREAM * FROM customers);
WITH ordersStream AS (SELECT STREAM * FROM orders);

INSERT INTO result
SELECT TABLE
    customersStream.name
    , ordersStream.item
 FROM
    ordersStream OUTER JOIN customersStream
        ON customersTable.id = ordersStream.customer_id;

An outer join can be seen as the union of left and right joins. It selects all records from the left and right side of the join regardless of a match happening:

Matching expression (ON)

By default, if no ON expression is provided, the join will be evaluated based on the equality of the _key facet. This means that the following queries are equivalent:

SELECT TABLE *
FROM customers JOIN orders;

SELECT TABLE *
FROM customers JOIN orders
    ON customers._key = orders._key;

When an expression is provided, however, there are limitations regarding what kind of expressions can be evaluated.

Currently, the following expression types are supported:

• Equality expressions using equality (=) with one table on each side:

  • customers.id = order.user_id

  • customers.id - 1 = order.user_id - 1

  • substr(customers.name, 5) = order.item

• Any boolean expression which references only one table:

  • len(customers.name) > 10

  • substr(customer.name,1) = "J"

  • len(customer.name) > 10 OR customer_key > 1

• Allowed expressions mixed together using an AND operator:

  • customers._key = order.user_id AND len(customers.name) > 10

  • len(customers.name) > 10 AND substr(customer.name,1) = "J"

  • substr(customers.name, 5) = order.item AND len(customer.name) > 10 OR customer_key > 1

Any expressions not following the rules above will be rejected:

• More than one table is referenced on each side of the equality operator

  • concat(customer.name, item.name) = "John"

  • customer._key - order.customer_id = 0

• a boolean expression not separated by an AND references more than one table:

  • customer._key = 1 OR customer._key = order.customer_id

Windowing: stream to stream joins (WITHIN )

When two streams are joined Lenses needs to know how far away in the past and in the future to look for a matching record.

This approach is called a “Sliding Window” and works like this:

Customers

Purchases

SELECT STREAM
     customers.name
    , orders.item
FROM
    customers LEFT JOIN orders WITHIN 5s
        ON customers.id = orders.customer_id
WITHIN 5s;

At t=10, when both the Computer and the Keyboard records arrive, only one customer can be found within the given time window (the specified window is 5s thus the window will be [10-5,10+5]s ).

This means that the following would be the result of running the query:

Note: John will not match the Keyboard purchase since t=20s is not within the window interval [10-5,10+5]s.

Non-windowed joins (stream to table and table to stream)

When streaming data, records can be produced at different rates and even out of order. This means that often a match may not be found because a record hasn’t arrived yet.

The following example shows an example of a join between a stream and a table where the arrival of the purchase information is made available before the customers’ information is.

(Notice that the purchase of a “Keyboard” by customer_id = 2 is produced before the record with the customer details is.)

Customers

Purchases

Running the following query:

WITH customersTable AS (SELECT TABLE * FROM customers);

SELECT STREAM
    customers.name
   , item.item
FROM
    orders LEFT JOIN customersTable  
        ON customers.id = orders.id

would result in the following:

If later, the record for customer_id = 2 is available:

a record would be emitted with the result now looking like the following:

Notice that “Keyboard” appears twice, once for the situation where the data is missing and another for when the data is made available.

This scenario will happen whenever a Stream is joined with a Table using a non-inner join.

Table/Stream joins compatibility table

The following table shows which combinations of table/stream joins are available:

Key decoder types

In order to evaluate a join between two sources, the key facet for both sources has to share the same initial format.

If formats are not the same the join can’t be evaluated. To address this issue, an intermediate topic can be created with the correct format using a STORE AS statement. This newly created topic can then be created as the new source.

Co-partitioning

In addition to the constraint aforementioned, when joining, it’s required that the partition number of both sources be the same.

When a mismatch is found, an additional step will be added to the join evaluation in order to guarantee an equal number of partitions between the two sources. This step will write the data from the source topic with a smaller count of partitions into an intermediate one.

This newly created topic will match the partition count of the source with the highest partition count.

In the topology view, this step will show up as a Repartition Node.

ON expressions and key change

Joining two topics is only possible if the two sources used in the join share the same key shape and decoder.

When an ON statement is specified, the original key facet will have to change so that it matches the expression provided in the ON statement. Lenses will do this calculation automatically. As a result, the key schema of the result will not be the same as either one of the sources. It will be a Lenses calculated object equivalent to the join expression specified in the query.

Nullability

As discussed when addressing join types, some values may have null values when non-inner joins are used.

Due to this fact, fields that may have null values will be typed as the union of null and their original type.

Joining more than 2 sources

Within the same query, joins may only be evaluated between two sources.

When a join between more than two sources is required, multiple queries can be combined using a WITH statement:

WITH customerOrders AS (
 SELECT TABLE
    customer.name
   , order.item,
   , order._key AS order_id
 FROM
    customers INNER JOIN orders
        ON orders.customer_id = customers.id
);

INSERT INTO target
SELECT TABLE *
FROM customerOrders INNER JOIN deliveryAddress
    ON customerOders.order_id = deliveryAddress.order_id;

Joining and Grouping

In order to group the results of a join, one just has to provide a GROUP BY expressions after a join expression is specified.

SET defaults.topic.autocreate=true;
WITH customersTable AS (SELECT TABLE * FROM customers);
WITH ordersStream AS (SELECT STREAM * FROM orders);

WITH joined AS (
    SELECT STREAM
        customersTable.name
        , ordersStream.item
     FROM
        ordersStream JOIN customersTable
            ON customersTable.id = ordersStream.customer_id
    GROUP BY customersTable.name;
);

Stream-Table/Table-Stream joins: table materialization

Purchases

WITH customersStream AS (SELECT TABLE * FROM customers);
WITH ordersStream AS (SELECT STREAM * FROM orders);

INSERT INTO result
SELECT TABLE
    customersStream.name
   , ordersStream.item
 FROM
    ordersStream OUTER JOIN customersStream
        ON customersTable.id = ordersStream.customer_id;

When a join between a table and a join is processed, lenses will, for each stream input (orders in the example above), look for a matching record on the specified table (customers).

Notice that the record with Frank’s purchase information is processed at t = 10s at which point Frank’s Customer information hasn’t yet been processed. This means that no match will be found for this record.

At t=20s however, the record with Frank’s customer information is processed; this will only trigger the emission of a new record if an Outer Join is used.

Filter optimizations

There are some cases where filter expressions can help optimize a query. A filter can be broken down into multiple steps so that some can be applied before the join node is evaluated. This type of optimization will reduce the number of records going into the join node and consequentially increase its speed.

For this reason, in some cases, filters will show up before the join in the topology node.

SQL Processors see data as an independent sequence of infinite events. An Event in this context is a datum of information; the smallest element of information that the underlying system uses to communicate. In Kafka’s case, this is a Kafka record/message.

Two parts of the Kafka record are relevant:

• Key

• Value

These are referred to as facets by the engine. These two components can hold any type of data and Kafka itself is agnostic on the actual storage format for either of these two fields. SQL Processors interpret records as (key, value) pairs, and it exposes ways to manipulate these pairs in several ways.

Queries are applications: SQL Processors

As mentioned above, queries that are meant to be run on streaming data are treated as stand-alone applications. These applications, in the context of the Lenses platform, are referred to as SQL Processors.

A SQL Processor encapsulates a specific Lenses SQL query, its details and everything else Lenses needs to be able to run the query continuously.

Schemas must be available for Structured Data

To support features like:

• Inference of output schemas

• Creation-time validation of input query

• Selections

• Expressions

Lenses SQL Engine Streaming mode needs to have up-to-date schema information for all structured topics that are used as input in a given query. In this context, structured means topics that are using complex storage formats like AVRO or JSON.

For the above query, for example, the purchases topic will need to have a value set to a structured format and a valid schema will need to already have been configured in Lenses. In such schema, fields itemId, price and quantity must be defined, the latter two being of a numerical type.

These requirements ensure the Engine will always be in a position to know what kind of data it will be working with, guaranteeing at the same time that all obvious errors are caught before a query is submitted.

M-N topologies

The UI allows us to visualise any SQL Processor out of the box. For the example:

This visualisation helps to highlight that the Lenses SQL fully supports M-N topologies.

What this means is that multiple input topics can be read at the same time, their data manipulated in different ways and then the corresponding results stored in several output topics, all as part of the same Processor’s topology.

This means that all processing can be done in one go, without having to split parts of a topology into different Processors (which could result in more data being stored and shuffled by Kafka).

Expressions in Lenses SQL

An expression is any part of a Lenses SQL query that can be evaluated to a concrete value (not to be confused with a record value).

In a query like the following:

CONCAT('a', 'b'), (1 + field1) and field2 are all expressions whose values will be _projected_ onto the output topic, whereas LENGTH(field2) > 5 is an expression whose values will be used to filter out input records.

SQL Processors and Kafka Streams

SQL Processors are built on top of Kafka Streams, and it enriches this tool with an implementation of Lenses SQL that fits well with the architecture and design of Kafka Streams. When executed, they run a Kafka Streams instance.

Consumer groups

Each SQL Processor has an application ID which uniquely identifies it within Lenses. The application ID is used as the Kafka Streams application ID which in turn becomes the underlying Kafka Consumer(s) group identifier.

Scaling

Scaling up or down the number of runners automatically adapts and rebalances the underlying Kafka Streams application in line with the Kafka group semantics.

The advantages of using Kafka Streams as the underlying technology for SQL Processors are several:

• Kafka Streams is an enterprise-ready, widely adopted and understood technology that integrates natively with Kafka

• Using consumer group semantics allows leveraging Kafka’s distribution of workload, fault tolerance and replication out of the box

Data as a flow of events: Streams

A stream is probably the most fundamental abstraction that SQL Processors provide, and it represents an unbounded sequence of independent events over a continuously changing dataset.

Let’s clarify the key terms in the above definition:

• event: an event, as explained earlier, is a datum, that is a (key, value) pair. In Kafka, it is a record.

• continuously changing dataset: the dataset is the totality of all data described by every event received so far. As such, it is changed every time a new event is received.

• unbounded: this means that the number of events changing the dataset is unknown and it could even be infinite

• independent: events don’t relate to each other and, in a stream, they are to be considered in isolation

The main implication of this is that stream transformations (e.g. operations that preserve the stream semantics) are stateless because the only thing they need to take into account is the single event being transformed. Most Projections fall within this category.

Stream Example

To illustrate the meaning of the above definition, imagine that the following two events are received by a stream:

Now, if the desired operation on this stream was to sum the values of all events with the same key (this is called an Aggregation), the result for "key1" would be 30, because each event is taken in isolation.

Finally, compare this behaviour with that of tables, as explained below, to get an intuition of how these two abstractions are related but different.

Stream syntax

Lenses SQL streaming supports reading a data source (e.g. a Kafka topic) into a stream by using SELECT STREAM.

The above example will create a stream that will emit an event for each record, including future ones.

Data as a snapshot of the state of the world: Tables

While a stream is useful to have visibility to every change in a dataset, sometimes it is necessary to hold a snapshot of the most current state of the dataset at any given time.

This is a familiar use-case for a database and the Streaming abstraction for this is aptly called table.

For each key, a table holds the latest version received of its value, which means that upon receiving events for keys that already have an associated value, such values will be overridden.

A table is sometimes referred to as a changelog stream, to highlight the fact that each event in the stream is interpreted as an update.

Given its nature, a table is intrinsically a stateful construct, because it needs to keep track of what it has already been seen. The main implication of this is that table transformations will consequently also be stateful, which in this context means that they will require local storage and data being copied.

Additionally, tables support delete semantics. An input event with a given key and a null value will be interpreted as a signal to delete the (key, value) pair from the table.

Finally, a table needs the key for all the input events to not be null. To avoid issues, tables will ignore and discard input events that have a null key.

Table example

To illustrate the above definition, imagine that the following two events are received by a table:

Now, if the desired operation on this table was to sum the values of all events with the same key (this is called an Aggregation), the result for key1 would be 20, because (key1, 20) is interpreted as an update.

Finally, compare this behaviour with that of streams, as explained above, to get an intuition of how these two abstractions are related but different.

Table syntax

Lenses SQL Streaming supports reading a data source (e.g. a Kafka topic) into a table by using SELECT TABLE.

The above example will create a table that will treat each event on input-topic, including future ones, as updates.

Tables and compacted topics

Given the semantics of tables, and the mechanics of how Kafka stores data, the Lenses SQL Streaming will set the cleanup.policy setting of every new topic that is created from a table to compact, unless explicitly specified otherwise.

What this means is that the data on the topic will be stored with a semantic more closely aligned to that of a table (in fact, tables in Kafka Streams use compacted topics internally). For further information regarding the implications of this, it is advisable to read the official Kafka Documentation about cleanup.policy.

The duality between streams and tables

Streams and tables have significantly different semantics and use cases, but one interesting observation is that are strongly related nonetheless.

This relationship is known as stream-table duality. It is described by the fact that every stream can be interpreted as a table, and similarly, a table can be interpreted as a stream.

• Stream as Table: A stream can be seen as the changelog of a table. Each event in the stream represents a state change in the table. As such, a table can always be reconstructed by replaying all events of a stream, in order.

• Table as Stream: A table can be seen as a snapshot, at a point in time, of the latest value received for each key in a stream. As such, a stream can always be reconstructed by iterating over each (Key, Value) pair and emitting it as an event.

To clarify the above duality, let’s use a chess game as an example.

On the left side of the above image, a chessboard at a specific point in time during a game is shown. This can be seen as a table where the key is a given piece and the value is its position. Also, on the right-hand side, there is the list of moves that culminated in the positioning described on the left; it should be obvious that this can be seen as a stream of events.

The idea formalised by the stream-table duality is that, as it should be clear from the above picture, we can always build a table from a stream (by applying all moves in order).

It is also always possible to build a stream from a table. In the case of the chess example, a stream could be made where each element represents the current state of a single piece (e.g. w: Q h3).

This duality is very important because it is actively used by Kafka (as well as several other storage technologies), for example, to replicate data and data stores and to guarantee fault tolerance. It is also used to translate table and stream nodes within different parts of a query.

SQL Processors and schemas: a proactive approach

One of the main goals of SQL Processors is to ensure that it uses all the information available to it when a SQL Processor is created to catch problems, suggest improvements and prevent errors. It’s more efficient and less frustrating to have an issue coming up during registration rather than at some unpredictable moment in the future, at runtime, possibly generating corrupted data.

SQL engine will actively check the following during the registration of a processor:

• Validation of all user inputs

• Query lexical correctness

• Query semantics correctness

• Existence of the input topics used within the query

• User permissions to all input and output topics

• Schema alignment between fields and topics used within the query

• Format alignment between data written and output topics, if the latter already exist

When all the above checks pass, the Engine will:

• Generate a SQL Processor able to execute the user’s query

• Generate and save valid schemas for all output topics to be created

• Monitor the processor and make such metrics available to Lenses

The Engine takes a principled and opinionated approach to schemas and typing information; what this means is that, for example, where there is no schema information for a given topic, that topic’s fields will not be available to the Engine, even if they are present in the data; also, if a field in a topic is a string, it will not be possible to use it as a number for example, without explicitly CASTing it.

The Engine’s approach allows it to support naming and reusing parts of a query multiple times. This can be achieved using the dedicated statement WITH.

The WITHs allow for whole sections of the query to be reused and manipulated independently by successive statements, and all this is done by maintaining schema and format alignment and correctness. The reason why this is useful is that it allows to specify queries that split their processing flow without having to redefine parts of the topology. This, in turn, means that less data needs to be read and written to Kafka, improving performance.

This is just an example of what SQL Processors can offer because of the design choices taken and the strict rules implemented at query registration.

A data stream is a sequence of events ordered by time. Each entry contains a timestamp component, which aligns it on the time axis.

Kafka provides the source for the data streams, and the Kafka message comes with the timestamp built in. This is used by the Push Engines by default. One thing to consider is that from a time perspective, the stream records can be out of order. Two Kafka records, R1 and R2 do not necessarily respect the rule: R1 timestamp is smaller than R2 timestamp.

Timestamps are required to perform time-dependent operations for streams - like aggregations and joins.

Timestamp semantics

A record timestamp value can have three distinct meanings. Kafka allows to configure a topic timestamp meaning via this log.message.timestamp.type setting. The two supported values are CreateTime and LogAppendTime.

Event time

When a record is created at source the producer is responsible for setting the timestamp for it. Kafka producer provides this automatically and this is aligned with the CreateTime the configuration mentioned earlier.

Ingestion time

At times, the data source timestamp is not available. When setting the topic timestamp type to LogAppendTime, the Kafka broker will attach the timestamp at the moment it writes it to the topic.

Processing time

The timestamp will be set to the time the record was read by the engine, ignoring any previously set timestamps.

Control the timestamp

Sometimes, when the data source is not under direct control, it might be that the record’s timestamp is actually embedded in the payload, either in the key or the value.

Lenses SQL Streaming allows to specify where to extract the timestamp from the record by using EVENTTIME BY.

where <selection> is a valid selection.

Here are a few examples on how to use the syntax to use the timestamp from the record value facet:

For those scenarios when the timestamp value lives within the record key, the syntax is similar:

Output timestamp

All records produced by the Lenses SQL Streaming will have a timestamp set and its value will be one of the following:

1. For direct transformations, where the output record is a straightforward transformation of the input, the input record timestamp will be used.

2. For aggregations, the timestamp of the latest input record being aggregated will be used.

3. In all other scenarios, the timestamp at which the output record is generated will be used.

Time windows

Some stream processing operations, like joins or aggregations, require distinct time boundaries which are called windows. For each time window, there is a start and an end, and as a result a duration. Performing aggregations over a time window, means only the records which fall within the time window boundaries are aggregated together. It might happen for the records to be out-of-order and arrive after the window end has passed, but they will be associated with the correct window.

Types

There are three-time windows to be used at the moment: hopping, tumbling and session.

Duration types

When defining a time window size, the following types are available:

Hopping window

These are fixed-size and overlapping windows. They are characterised by duration and the hop interval. The hop interval specifies how far a window moves forward in time relative to the previous window.

Since the windows can overlap, a record can be associated with more than one window.

Use this syntax to define a hopping window:

Tumbling window

They are a particularisation of hopping windows, where the duration and hop interval are equal. This means that two windows can never overlap, therefore a record can only be associated with one window.

Duration time takes the same unit types as described earlier for hopping windows.

Session window

Unlike the other two window types, this window size is dynamic and driven by the data. Similar to tumbling windows, these are non-overlapping windows.

A session window is defined as a period of activity separated by a specified gap of inactivity. Any records with timestamps that occur within the boundaries of the inactivity interval are considered part of the existing sessions. When a record arrives and its timestamp is outside of the session gap, a new session window is created and the record will belong to that.

A new session window starts if the last record that arrived is further back in time than the specified inactivity gap. Additionally, different session windows might be merged into a single one if an event is received that falls in between two existing windows, and the resulting windows would then overlap.

To define a session window the following syntax should be used:

The inactivity interval can take the time unit type seen earlier for the hopping window.

Session windows are tracked on a per-key basis. This means windows for different keys will likely have different durations. Even for the same key, the window duration can vary.

User behaviour analysis is an example of when to use session windows. They allow metrics like counting user visits, customer conversion funnel or event flows.

Late arrival

It is quite common to see records belonging to one window arriving late, that is after the window end time has passed. To accept these records the notion of a grace period is supported. This means that if a record timestamp falls within a window W and it arrives within W + G (where G is the grace interval) then the record will be processed and the aggregations or joins will update. If, however, the record comes after the grace period then it is discarded.

To control the grace interval use this syntax:

The default grace period is 24 hours. Until the grace period elapses, the window is not actually closed.

Lenses has two SQL engines to allow users to explore and process streaming data:

1. SQL Snapshot - This is point in time SQL engine powering the Explore and SQL Studio for debugging and exploration of data.

2. SQL Processors (Streaming) - These are long-running applications, configured, deployed and managed by Lenses to perform joins, aggregations, data conversion and more.

When running queries against Kafka, Snapshot Engine enables you to access the record metadata through the special _meta facet.

These are the available meta fields:

Select all the meta fields

The following query will select all the meta fields listed above:

SELECT _meta.* FROM topic

View headers

To view the value of a specific header you can run:

SELECT HEADERASSTRING("User") as user
FROM trips
LIMIT 100

Filter on record timestamp

SELECT ...
    FROM topic
WHERE _meta.timestamp > YESTERDAY()

Filter on table partition

To read records from a specific partition, the following query can be used:

SELECT ...
    FROM topic
WHERE _meta.partition = 1
    OR _meta.partition = 8

Search for a record on a specific offset

Here is the query to use when the record offset and partition are known:

SELECT ...
    FROM topic
WHERE _meta.partition = 2
    AND _meta.offset = 8
LIMIT 1

Get the latest N records per partition

This query will get the latest 100 records per partition (assuming the topic is not compacted):

SELECT ...
    FROM topic
WHERE _meta.offset >= LAST_OFFSET() - 100

This instead will get the latest 100 records for a given partition (again assuming the topic is not compacted):

SELECT ...
    FROM topic
WHERE _meta.offset >= LAST_OFFSET() - 100 
    AND _meta.partition = 2

A projection represents the ability to project a given input value onto a target location in an output record. Projections are the main building block of SELECT statements.

A projection is composed of several parts. Projections have a source section that allows one to select a specific value and to ensure that it will be present in the output, in the desired structure and with the desired name, as described by the target section.

In the below query:

These are projections:

• CONCAT(‘a’, ‘b’) as result1

• field4

• (1 + field1) as _key.a

• _key.field2 as result3

• 5 + 7 as constantField

• CASE … as who_is_it

It is worth highlighting that projections themselves are stateless. While the calculation of the source value could be stateful (depending on the type of expression being evaluated, the act of making the value available in the output is not a stateful operation.

Syntax

The precise syntax of a projection is:

In the above, [] indicates optional sections and | is to be read as OR.

• Source:

  • <expression>: the source section must consist of a valid SQL expression, which will be evaluated to the value to be projected.

• Target: this section is optional. When missing, the output target facet will be _value and the <alias> will be defaulted to the string representation of the source expression.

  • as: this is the Lenses SQL keyword that makes specifying the target of the projection explicit.

  • [<facet>.]<alias>]|[<facet>]: this nested section, which must be specified whenever as is used, is composed of two mutually exclusive sub-sections.

    • [<facet>.]<alias>]: this optional section is a string that will be used as the field name in the output record. Optionally, the output _facet_ where the field will be projected (e.g. _key or _value) can be also specified.

    • [<facet>]: this optional section specifies that the result of the projection will make up the whole of the indicated facet of the output record (e.g. the whole key or the whole value).

The above syntax highlights something important about the relationship between projections and facets: a projection can have a source facet and will always have a target facet and while they are related, they might not always be the same.

In the above example query:

• field4 is a projection from value facet to value facet

• (1 + field1) as _key.a is a projection from the value facet to the key facet

• _key.field2 as result3 is a projection from the key facet to the value facet

• 5 + 7 as constantField is a projection to value facet but with no source facet, as the expression does not depend on the input record

This makes SQL projections very expressive, but it also means that attention needs to be paid when facets are manipulated explicitly. Details, edge cases and implications of this will be discussed below.

Source of a Projection

Projection from a _value field

This is the default type of projection when a field is selected within the source expression. Unless otherwise specified, the source facet of a projection will always be _value.

In light of the above, the following query contains only projections that read from the _value facet of the source:

Also, notice that field1 and _value.field1 as aliased are reading the exact same input field and in the former case the _value facet is implicit.

Projection from a key field

A projection can also access a selected field on the key facet.

The above query contains only projections that read from the _key facet of the source.

This kind of projection behaves exactly the same as any other projection, but because of specific interactions with other mechanics in Lenses SQL Engine Streaming mode, they can’t be used in the same query with Aggregations or Joins.

Projection from whole facets

All examples of projections described until now focused on selecting a field from either the key or the value of an input record. However, a projection can also read a whole facet and project it to the output.

In the above query, there are two projections:

• _key as old-key: This is projecting the whole key of input-topic onto a field old-key on target-topic

• _value as old-value: This is projecting the whole value of input-topic onto a field old-value on target-topic

For more details about the rules around using aliases (as done in the above example).

This can be useful when the input source uses a primitive storage format for either one or both facets, but it is desirable to map such facets to named fields within a more complex output structure, as could be the case in the above query. This said projections from whole facets are supported for all storage formats, not only the primitive ones.

Mix and match

As it should be clear from all the examples on this page so far, projections can be freely mixed within a single SELECT statement; the same query can have many projections, some of which could be reading from the key of the input record, some others from the value and yet others returning literal constants.

Lenses SQL is designed to support this mixed usage and to calculate the appropriate resulting structure given the schemas of all the projections’ inputs.

Wildcard projections

Lenses SQL assigns a special meaning to * when used as a projection.

Unqualified Wildcard projection

When * is used without any further qualification, it is interpreted as an instruction to project all fields from _key_ and _value_ to the output.

The result of this query is that target-topic will have exactly the same fields, schema and data than input-topic.

Qualified wildcard projection

When * is explicitly qualified, the meaning becomes more precise and it will limit the fields to be selected to only the ones belonging to the qualified source (and optionally facet).

The above shows how a qualified wildcard projection can be used to target all the fields of a specific source. Additionally, a qualified wildcard can be used in addition to other normal projections (e.g. i2.field1).

Target of a Projection

The target of a projection is the location within the output record where the result of the projection’s expression is going to be mapped. As previously mentioned, Lenses SQL uses the keyword as to explicitly control this.

Using as, it is possible to:

• Assign an alias to the projected field in the result. For example, field1 as aliased-field1 is reading field1 and projecting its value to a field called aliased-field1. Notice that, as no facet information is specified, this will be targeting the value of the output record.

• Project directly to nested fields within structures

The above query will result in a field x that is a structure that contains two fields a and b.

• Control the facet to which the field is projected. For example, field1 as _key.field1 is reading field1 (from the value facet) and projecting to a field with the same name on the key of the output. Depending on the source being projected, doing this might have important implications.

• Project over the whole target value or key For example, field1 as _value is reading field1 and projecting it over the whole value of the output. One important thing about this is that Lenses SQL allows only one projection of this kind per facet per query. What this means is that the following query would be invalid, because only one projection can target _value within the same query:

Rules for nested aliases

In order to avoid potential errors, Lenses defines the following rules for defining aliases:

• Alias can’t add new fields to non-struct properties.

  • e.g: an_int AS foo.bar, field1 AS foo.bar.field1 (since bar will be an INT we can’t define a property field1 under it)

• Aliases cannot override previously defined properties.

  • e.g: field1 AS foo.bar.field1, field2 as foo.bar (setting foo.bar with the contents of field2 would override the value of field1)

• Fields cannot have duplicated names.

  • e.g: field1 AS foo.bar, field2 as foo.bar (setting foo.bar with the contents of field2 would override it’s previous content)

Implications of projecting on key

Projecting on Key is a feature that can be useful in situations where it is desirable to quickly change the key of a Table or a Stream, maybe in preparation for further operations (e.g. joins etc…). This feature is sometimes referred to as re-keying within the industry.

However, one important implication of using this feature is that Kafka uses the key to determine in what partition a record must be stored; by changing the key of the record, the resulting partitioning of the output topic might differ from one of the input topics. While there is nothing wrong with this, it is something that must be understood clearly when using this feature.

Filtering

Sometimes it is desirable to limit the input records to be projected based on some predicate.

For example, we might want to project field1 and field2 of input-topic onto output-topic, but only if field3 contains a specific value.

This is the WHERE clause is used: to filter the input dataset by some predicate, applying the rest of the query only to records that match the predicate.

The syntax for this clause is simply WHERE <expression>, where <expression> is a valid arbitrarily nested Lenses SQL boolean expression.

Projections and Storage Formats

Projections have a close relationship with the storage format of their target.

By default, if a query contains more than one projection for the same facet, then that facet’s storage format will be a structure (which type of structure exactly depends on other factors that are not relevant here).

The above query will make target-topic’s value storage format a structure (e.g. AVRO or JSON) with two fields named: result1 and field2. The storage format for target-topic’s key will be the same as the input-topic’s, as there are no projections targeting that facet.

The storage format of the output can be explicitly changed by a projection, however. This will often be the case when a projection on a whole facet is used. Consider the following query:

In this case, target-topic’s value storage format will still be a structure, but its key will depend on field2’s schema. For example, if field2 is a string, then target-topic’s key will be changed to STRING (assuming it was not STRING already). The same behavior applies to the _value_ facet.

One example where this can be relevant is when a projection is used to map the result of a single field in the target topic. Consider the following query:

This query will project field1 on the whole value facet, and this will result in a change of storage format as well. This behaviour is quite common when a single projection is used, because more often than not the desired output in such a scenario will be the content of the field rather than a structure with a single field.

Aggregations are stateful transformations that allow to grouping of an unbounded set of inputs into sub-sets and then aggregate each of these sub-sets into a single output; the reason why they are stateful is because they need to maintain the current state of computation between the application of each input.

To group a given input dataset into sub-sets, a key function needs to be specified; the result of applying this key function to an input record will be used as a discriminator (sometimes called a pivot) to determine in what sub-set each input record is to be bucketed.

The specific transformation that each aggregation performs is described by the Aggregated Functions used in the input query.

Aggregations match table semantics

Notice that the behaviour described above is precisely what a Table does. For any given key, there is the state will continuously be updated as new events with the given key are received. In the case of Aggregations, new events are represented by input records in the original dataset that will map to a given key, therefore ending up in a bucket or another.

Whenever Aggregations are used, the result will be a Table. Each entry will have the key set to the grouping discriminator, and the value set to the current state of computation for all input records matching the key.

Syntax

The complete syntax for aggregations is:

The specific syntactical elements of the above are:

• (STREAM | TABLE): specifies if the <source> is to be interpreted as a stream or a table.

• <aggregated projection1>: a projection is aggregated when its source contains an Aggregated Function (e.g. COUNT(*) as x, CAST(COUNT(*) as STRING) as stringed).

• [, aggregated projection2] ... [, aggregated projectionN]: a query can contain any number of additional aggregated projections after the first mandatory one.

• [, projection1] ... [, projectionM]: a query can contain any number of common, non-aggregated, projections. Streaming only supports full GROUP BY mode. This means that fields that are not part of the GROUP BY clause cannot be referenced by non-aggregated projections.

• <source>: a normal source, like a topic or the result of a WITH statement.

• [WINDOW BY <window description>]: this optional section can only be specified if STREAM is used. It allows us to describe the windowing that will be applied to the aggregation.

• GROUP BY <expression>: the result of evaluating <expression> will be used to divide the input values into different groups. These groups will be the input for each aggregated projection specified. The <expression>’s result will become the key for the table resulting from the query.

Specific rules for aggregated projections

Most of the rules and syntax described for Projections apply to aggregated projections as well, but there are some additional syntactical rules due to the specific nature of Aggregations.

• Aliasing rules mostly work the same, but it is not possible to project on the key facet; COUNT(*) as _key.a or SUM(x) as _key are therefore not allowed.

• At least one aggregated projection must be specified in the query.

• Projections using an unqualified key facet as the source are not allowed. _key.a or COUNT(_key.b) are forbidden because _key is unqualified, but <source>._key.a and COUNT(<source>._key.b) are supported.

Grouping and storage format for key

As previously mentioned, the GROUP BY is used to determine the key of the query’s result; the above query will group all records in input-topic by the value of field1 in each record, and target-topic’s key will be the schema of field1.

Just like in the case of the Projections, the Streaming mode takes an opinionated approach here and will simplify the result schema and Storage Format in the case of single field structures.

In the case above, assuming for example that field1 is an integer, target-topic’s key will not be a structure with a single integer field1 field, but rather just the value field1; the resulting storage format is going to be INT, and the label field1 will be just dropped.

In case the above behaviour is not desirable, specifying an explicit alias will allow us to override it.

This will result in target-topic’s key is a structure with a field keep_me, with the same schema as field1. The corresponding Storage Format will match the input format for input-topic, AVRO or JSON.

Semantics

An example will help clarify how aggregations work, as well as how they behave depending on the semantics of the input dataset they are being applied to.

Example scenario

Assume that we have a Kafka topic (gaming-sessions) containing these records:

What this data describes is a series of gaming sessions, performed by a player. For each gaming session, the player (used as Key), the points achieved, and the country where the game took place.

Aggregating a Stream

Let’s now assume that what we want to calculate is the total points achieved by players in a given country, as well as the average points per game. One way to achieve the desired behaviour is to build a Stream from the input topic. Remember that this means that each event will be considered in isolation.

Explanations for each element of this syntax can be found below, but very briefly, this builds a Stream from gaming-sessions, grouping all events by country (e.g. all records with the same country will be aggregated together) and finally calculating the total (total_points) and the average (average_points) of all points for a given group.

The final result in target-topic will be (disregarding intermediate events):

The results are calculated from the totality of the input results because in a Stream, each event is independent and unrelated to any other.

Aggregating a Table

We now want to calculate something similar to what we obtained before, but we want to keep track only of the last session played by a player, as it might give us a better snapshot of both the performances and locations of players worldwide. The statistics we want to gather are the same as before: total and average of points per country.

The way to achieve the above requirement is simply by reading gaming-sessions into a Table, rather than a Stream, and aggregating it.

The final result in target-topic will be (disregarding intermediate events):

Compare this with the behaviour from the previous scenario; the key difference is that the value for uk includes only willy and noel, and that’s because the last event moved billy to the spain bucket, removing all data regarding him from his original group.

Aggregation functions for Tables

The previous section described the behaviour of aggregations when applied to Tables, and highlighted how aggregations not only need to be able to sum the latest values received to the current state of a group but also need to be able to subtract an obsolete value that might have just been assigned to a new group. As we saw above, it is easy to do this in the case of SUM and AVG.

However, consider what would happen if we wanted to add new statistics to the ones calculated above: the maximum points achieved by a player in a given country.

In the Stream scenario, this can be achieved by simply adding MAXK(points,1) as max_points to the query.

In the Table scenario, however, things are different. We know that the final event moves billy from uk to spain, so we need to subtract from uk all information related to billy. In case of SUM and AVG that’s possible because subtracting billy’s points to the current value of the aggregation will return the correct result.

But that’s not possible for MAXK. MAXK(points, 1) only keeps track of 1 value, the highest seen so far, and if that’s removed, what value should take its place? The aggregation function cannot inspect the entire topic data to search for the correct answer. The state the aggregation function has access to is that single number, which now is invalid.

This problem explains why some aggregated functions can be used on Streams and Tables (e.g. SUM), while others can be used only on Streams (e.g. MAXK).

The key factor is usually whether a hypothetical subtraction operation would need access to all previous inputs to calculate its new value (like MAXK) or just the aggregated state (like SUM).

Windowed aggregations

A common scenario that arises in the context of aggregations is the idea of adding a time dimension to the grouping logic expressed in the query. For example, one might want to group all input records by a given field that were received within 1 hour of each other.

To express the above Lenses SQL Streaming supports windowed aggregations, by adding a WINDOW BY clause to the query. Given their semantics, tables cannot be aggregated using a window, because it would not make sense. A table represents the latest_ state of a set of (Key, Value) pairs, not a series of events interspersed over a time continuum. Thus trying to window them is not a sensible operation.

Filtering aggregated queries

Filtering the input into aggregated queries is similar to filtering non-aggregated ones. When using a WHERE <expression> statement, where <expression> is a valid SQL boolean expression, all records that do not match the predicate will be left out.

However, aggregated functions add a further dimension to what it might be desirable to filter.

We might be interested in filtering based on some conditions of the groups themselves; for example, we might want to count all input records that have a given value offield1, but only if the total is greater than 3. In this case, WHERE would not help, because it has no access to the groups nor to the results of the aggregated projections. The below query is what is needed.

The above query uses the HAVING clause to express a filter at a grouping level. Using this feature it is possible to express a predicate on the result of aggregated projections and filter out the output records that do not satisfy it.

With Lateral Joins you can combine a data source with any array expression. As a result, you will get a new data source, where every record of the original one will be joined with the values of the lateral array expression.

Assume you have a source where elements is an array field:

Then a Lateral Join of source with elements is a new table, where every record of source will be joined with all the single items of the value of elements for that record:

In this way, the single elements of the array become available and can be used as a normal field in the query.

Syntax

A query using lateral joins looks like a regular query apart from the definition of its source:

• projection: as in a single-table select, all the fields from <source> will be available in the projection. In addition to that, the special field <lateralAlias> will be available.

• source: the source of data. Note: it is not possible to specify a normal join as a source of a lateral join. This limitation will be removed in the future.

• lateralArrayExpression: any expression that evaluates to an array. Fields <source> are available for defining this expression.

• filterExpression: a filter expression specifying which records should be filtered.

Single Lateral Joins

Assume you have a topic batched_readings populated with the following records:

batched_readings

As you can see, readings is a field containing arrays of integers.

We define a processor like this:

The processor will emil the following records:

Things to notice:

• We used the aliased lateral expression reading both in the projection and in the WHERE.

• The _key for each emitted record is the one of the original record. As usual you can change this behavior projecting on the key with a projection like expression AS _key.

• batched_readings records with keys a and b have been split into multiple records. That’s because they contain multiple readings greater than 90.

• Record d disappeared, because it has no readings greater than 90

Multiple Lateral Joins

It is possible to use multiple LATERAL joins in the same FROM clause.

Assume you have a topic batched_nested_readings populated with the following records:

batched_readings

Notice how nested_readings contains arrays of arrays of integers.

To get the same results of the previous example, we use a first lateral join to unpack the first level of nested_readings into an array that we call readings. We then define a second lateral join on readings to extract the single values:

Complex Lateral expressions

In the previous example we used a simple field as the <lateralArrayExpression>. In the section we will see how any array expression can be used for it.

Assume you have a topic day_night_readings populated with the following records:

day_night_readings

We can make use of Array Functions to lateral join day_night_readings on the concatenation of the two readings fields:

Null values are used as a way to express a value that isn’t yet known.

Null values can be found in the data present in existing sources, or they can be the product of joining data using non-inner joins.

The schema of nullable types is represented as a union of the field type and a null value:

Using nullable types

Working with null values can create situations where it’s not clear what the outcome of an operation is.

One example of this would be the following:

Looking at the first two expressions, one may be tempted to solve the problem above by saying “Null is 1 when multiplying and 0 when summing” meaning the following would be the evaluation result:

Rewriting the third expression applying the distributive property of multiplication however, shows that the rule creates inconsistencies:

With the intent of avoiding scenarios like the above where a computation may have different results based on the evaluation approach taken, most operations in lenses do not allow operations to use nullable types.

Address nullability

Lenses provides the following tools to address nullability in a flexible way:

COALESCE

• Coalesce: A function that allows specifying a list of fields to be tried until the first non-null value is found.

  • Note: the coalesce function won’t verify if a non-nullable field is provided so an error may still be thrown if all the provided fields are null

  • e.g:COALESCE(nullable_fieldA, nullable_fieldB, 0)

AS_NON_NULLABLE

• AS_NON_NULLABLE: a function that changes the type of a property from nullable to non-nullable.

  • Note: This function is unsafe and will throw an error if a null value is passed. It should only be used if there’s a guarantee that the value won’t ever be null (for instance if used in a CASE branch where the null case has been previously handled or if the data has previously been filtered and the null values removed).

  • e.g: AS_NON_NULLABLE(nullable_field)

AS_NON_NULLABLE and CASE

• AS_NON_NULLABLE with CASE: A type-checked construct equivalent to using coalesce:

  • e.g:

AS_NULLABLE

The AS_NULLABLE function is the inverse transformation of the AS_NON_NULLABLE version. This function allows a non-nullable field type to be transformed into a nullable type. It can be used to insert data into existing topics where the schema of the target field is nullable.