Readers

This page explains how to set up reader operators, for kdb Insights Enterprise pipelines, using the Web Interface.

Reader operators enables both streaming and batch data to be ingested into kdb Insights Enterprise.

The pipeline builder uses a drag-and-drop interface to link together operators within a pipeline. For details refer to the Building a pipeline guide.

The following readers are available:

• Callback

• Amazon S3

• Expression

• Google Cloud Storage

• HTTP

• Kafka

• kdb Insights Database

• kdb Insights Stream

• Microsoft Azure Storage

• MQTT

• Parquet

• PostgreSQL

• SQL Server

• Upload

Callback

Use this reader to receives data from a callback from the local process or over IPC.

Example: Publishing from a tickerplant

A tickerplant is a message bus that typically sends data to subscribers in the form of (table; data). In this example, we subscribe to a tickerplant with a trades and a quotes table with independent callback readers.

1. Drag out a callback reader and set the name to be Callback name to trade.

2. Drag out a second callback reader and set the name to be Callback name to quote.

3. Setup the following global code, for establishing a connection to the tickerplant and dispatching to the appropriate callback function based on the table name.

   q

   Copy

       //Insert tickerplant connection setup code here
       // ...
   
       // Receives updates from a ticketplant
       // @param t {symbol} New table
       // @param x {table} Table data
       upd:{[t;x]
           $[`trades ~ t; trade x; quote x]
           };             

   

Expression

This reader runs a kdb+ expression.

The expression reader executes a snippet of kdb+ code and passes the result into the pipeline. This operator is mostly used for testing pipelines with mock data.

Required Parameters:

Optional Parameters:

The following screenshot shows the Trigger Read properties displayed when Trigger Mode is set to Timer.

Refer to Pull Trigger Read Options for more details.

Example: Generating data

The code snippet below generates a sample table with 2000 rows for testing:

n:2000;
([] date:n?(reverse .z.d-1+til 10);
  instance:n?`inst1`inst2`inst3`inst4;
  sym:n?`USD`EUR`GBP`JPY;
  num:n?10)
                    

import pandas as pd
import numpy as np
from datetime import datetime,timedelta

def gen():
    n=2000
    dates = [datetime.today().date() - timedelta(10-i,0,0) for i in range(10)]
    return pd.DataFrame({'date' : np.random.choice(dates,n),
              'instance' : np.random.choice(["inst1","inst2","inst3","inst4"],n),
              'sym' : np.random.choice(["EUR","USD","GBP","JPY"],n),
              'num' : np.random.randint(0,10,n)})

                    

Google Cloud Storage

This reads an object from Google Cloud Storage.

There are two versions of the reader:

• Version 1: The original and default version.

• Version 2: Supports parallel file downloads, significantly increasing throughput during processing. This is the recommended option.

To change the operator version, click Select Version and select either Version 1 or Version 2.

Click the Version 2 or Version 1 tab below to view the operator configuration options available for each version.

Required Parameters:

Optional Parameters:

Google Cloud Storage Authentication

Google Cloud Storage Authentication uses Kurl for credential validation.

Environment variable authentication

To setup authentication using an environment variable, set GOOGLE_STORAGE_TOKEN in Google's gcloud CLI

gcloud auth print-access-token

                    

Kubernetes secrets for authentication

A Kubernetes secret may be used to authenticate files. This secret needs to be created in the same namespace as the kdb Insights Enterprise install. The name of that secret is then used in the Kubernetes Secret field when configuring the reader.

To create a Kubernetes secret with Google Cloud credentials:

kubectl create secret generic SECRET_NAME DATA
                    

HTTP

This reader requests data from an HTTP(S) endpoint.

Required Parameters:

Optional Parameters:

Refer to Pull Trigger Read Options for more details.

Advanced Parameters:

Kafka

This reader consumes data from a Kafka topic.

Kafka is a scalable streaming message queue for processing streaming data. Refer to the Kafka ingestion example for details about setting ingesting from Kafka.

Required Parameters:

Optional Parameters:

Advanced Parameters:

kdb Insights Database

Use this reader to read from a kdb Insights Database.

The configuration options vary depending on the Query Mode you select.

• Get Data API

• UDA

Get Data API

The settings for reading from kdb Insights Database using Get Data API are described below.

1. Enter values for the following required parameters.

   Parameter	Description
   Table Name	The name of the table data is to be retrieved from.
   Start Date	The inclusive start time of the period of interest.
   End Date	The exclusive end time of the period of interest.

2. To add optional filters, click + Add Filter and click Add Function to select one of the filter options.

    

   Additional options are displayed depending on the filter option selected.

    

   Filter option	Description
   filter	A function that returns true for rows to include.
   filter by label	When multiple databases each have a table matching the table name, filtering by label can select a subset of those.
   sort	A column to sort by (ascending).
   define aggregation	Apply an aggregation over a column for the rows matching the query, assigning the result to a new column.
   group aggregation by	Group the result of the query by the specified column before running the aggregation.

3. Optional Parameters:

   name	description	default
   Read Mode	Configures how the node is triggered. This can be one of Once, API, Timer	Once
   Period	This field, displayed when the Read Mode is set to Timer, specifies the period to trigger the read on.	1D00:00:00
   Set Start Time	This field, displayed when the Read Mode is set to Timer, sets the time of the initial read. When unchecked the read triggers on execution. When checked a Start Time parameter is displayed.	Unchecked
   Start Time	The time the initial read is triggered.	Today

Refer to Pull Trigger Read Options for more details.

1. Click Advanced to optionally enter values for the following.

    

   Advanced Parameter	Description	Default
   Retry Attempts	Maximum number of retry attempts allowed.	5
   Retry Wait Time	Wait time between retry attempts.	2 seconds

UDA

The settings for reading from kdb Insights Database using UDAs are described below. See UDAs for details on how these analytics are created and loaded to kdb Insights Enterprise.

1. Select a User Defined Analytic(UDA) from the list of available UDAs.

2. Optional Parameters:

   name	description	default
   Read Mode	Configures how the node is triggered. This can be one of Once, API, Timer	Once
   Period	This field, displayed when the Read Mode is set to Timer, specifies the period to trigger the read on.	1D00:00:00
   Set Start Time	This field, displayed when the Read Mode is set to Timer, sets the time of the initial read. When unchecked the read triggers on execution. When checked a Start Time parameter is displayed.	Unchecked
   Start Time	The time the initial read is triggered.	Today

Refer to Pull Trigger Read Options for more details.

1. Click Advanced to optionally enter values for the following.

   Advanced Parameter	Description	Default
   Retry Attempts	Maximum number of retry attempts allowed.	5
   Retry Wait Time	Wait time between retry attempts.	2 seconds

MQTT

Use this reader to subscribe to an MQTT topic.

Read messages from an MQTT broker. MQTT is a lightweight, publish-subscribe, machine to machine network protocol for message queuing service. It is designed for connections with remote locations that have devices with resource constraints or limited network bandwidth.

Required Parameters:

Optional Parameters:

Microsoft Azure Storage

This reads an object from Microsoft Azure Storage. There are two versions of the reader:

• Version 1: The original and default version.

• Version 2: Supports parallel file downloads, significantly increasing throughput during processing. This is the recommended version

To change the operator version, click Select Version and select either Version 1 or Version 2.

Click the Version 2 or Version 1 tab below to view the operator configuration options available for each version.

Required Parameters:

Optional Parameters:

Microsoft Azure Authentication

Microsoft Azure Storage Authentication uses Kurl for credential validation.

Environment variable Authentication

To set up authentication with an environment variable, set AZURE_STORAGE_ACCOUNT to the name of the storage account to read from, and AZURE_STORAGE_SHARED_KEY to the one of the keys of the account. Additional details on shared keys usage available here.

Learn more about Shared Keys.

Kubernetes secrets for authentication

A Kubernetes secret may be used to authenticate files. This secret needs to be created in the same namespace as the kdb Insights Enterprise install. The name of that secret is then used in the Kubernetes Secret field when configuring the reader.

To create a Kubernetes secret with Azure credentials:

kubectl create secret generic --from-literal=token="$AZURE_STORAGE_TOKEN" ms-creds
                    

Learn more about Kubernetes Secrets.

kdb Insights Stream

This reads data from a kdb Insights Stream.

Insights streams

Optional Parameters:

To read data from a stream, an assembly must be provided. This is done prior to pipeline deploy time. Visit the Settings tab in the pipeline editor and fill in the Assembly Integration field with the assembly name and topic name.

Amazon S3

This reads an object from Amazon S3.

There are two versions of the reader:

• Version 1: The original and default version.

• Version 2: Supports parallel file downloads, significantly increasing throughput during processing. This is the recommended version.

To change the operator version, click Select Version to select either Version 1 or Version 2.

Click the Version 2 or Version 1 tab below to view the operator configuration options available for each version.

Required Parameters:

Optional Parameters:

Amazon S3 Authentication

To access private buckets or files, a Kubernetes secret needs to be created that contains valid AWS credentials. This secret needs to be created in the same namespace as the kdb Insights Enterprise install. The name of that secret is then used in the Kubernetes Secret field when configuring the reader.

To create a Kubernetes secret containing AWS credentials:

kubectl create secret generic --from-file=credentials=<path/to/.aws/credentials> <secret-name>
                    

Where <path/to/.aws/credentials> is the path to an AWS credentials file, and <secret-name> is the name of the Kubernetes secret to create.

Note that this only needs to be done once, and each subsequent usage of the Amazon S3 reader can re-use the same Kubernetes secret.

Parquet

This reads a parquet file.

Required Parameters:

Optional Parameters:

To configure authentication for each cloud provider, see the following:

• Amazon S3 Authentication

• Microsoft Azure Authentication

• Google Cloud Storage Authentication

Upload

This reads data supplied through an HTTP endpoint.

Required Parameters:

Example

Once the pipeline is deployed you can upload files using HTTP by editing the template curl command:

curl -X POST "https://insights.kx.com/streamprocessor/upload/{Endpoint Name}?table=z&finish=true" \
 --header "Content-Type: application/octet-stream" \
 --header "Authorization: Bearer $INSIGHTS_TOKEN" \
 --data-binary "@my/file.csv"

PostgreSQL

This executes a query on a PostgreSQL database.

Required Parameters:

Optional Parameters:

Refer to Pull Trigger Read Options for more details.

SQL Server

This executes a query on a SQL Server database.

Required Parameters:

Optional Parameters:

Refer to Pull Trigger Read Options for more details.

Further reading

• Building pipelines

• Import Wizard