CREATE DATA SINK - Kinetica Docs

Creates a new data sink, which contains the location and connection information for a data consumer that is generally external to Kinetica. A data sink serves as an authentication & access mechanism to the remote resource.

CREATE DATA SINK Syntax

CREATE [OR REPLACE] [EXTERNAL] DATA SINK [<schema name>.]<data sink name>
LOCATION = '<consumer>://<target>[:<port>]'
[WITH OPTIONS (<option name> = '<option value>'[,...])]

A data sink can be referenced in an EXPORT … INTO call (for exporting local table data to a remote system) or a CREATE STREAM call (for streaming local table data to a remote system).

The data sink will be validated upon creation, by default, and will fail to be created if an authorized connection cannot be established.

The following data sink consumers are supported:

Azure (Microsoft blob storage)
GCS (Google Cloud Storage)
HDFS (Apache Hadoop Distributed File System)
JDBC (Java Database Connectivity, using a user-supplied driver or one of the drivers on the supported list)
Kafka (Apache Kafka streaming feed)
Local (Table within the same Kinetica instance)
S3 (Amazon S3 Bucket)
Webhook (HTTP/HTTPS)

Kafka data sinks will be validated upon creation, by default, and will fail to be created if an authorized connection cannot be established.

For consumer-specific syntax, see Consumer-Specific Syntax. For consumer-specific examples, see Creating Data Sinks.

Parameters

OR REPLACE

Any existing data sink with the same name will be dropped before creating this one

EXTERNAL

Optional keyword for clarity

Name of the schema that will contain the created data sink; if no schema is specified, the data sink will be created in the user’s default schema

Name of the data sink to create; must adhere to the supported naming criteria

Consumer of the data sinkSupported consumers include:

Consumer	Description
`AZURE`	Microsoft Azure blob storage
`GCS`	Google Cloud Storage
`HDFS`	Apache Hadoop Distributed File System
`HTTP`	Unsecured webhook
`HTTPS`	Secured webhook
`JDBC`	JDBC connection, where `LOCATION` is the JDBC URL. `LOCATION = ‘jdbc:postgresql://example.com:5432/mydb’` See the supported list for the full list of supported drivers, or specify one with `JDBC_DRIVER_JAR_PATH` and `JDBC_DRIVER_CLASS_NAME`.
`KAFKA`	Apache Kafka broker
`TABLE`	Local Kinetica table residing in the same database as the data sink: `LOCATION = ‘TABLE://example.employee_backup’`
`S3`	Amazon S3 bucket

Target to connect to via data sink; the following consumers have a default host target, if target is left blank:

Consumer	Default Host
Azure	`<storage_account_name>.blob.core.windows.net`
GCS	`storage.googleapis.com`
S3	`<region>.amazonaws.com`

For local Kinetica tables, target is the name of the name of the target table:

[<target table schema name>].<target table name>

<port>

Port to connect to via the data sink, if applicable

WITH OPTIONS

Optional indicator that a comma-delimited list of connection option/value assignments will follow. See Options for the full list of options.

Examples

To create a data sink, kin_dsink, that connects to an Apache Kafka broker:

CREATE DATA SINK Example

CREATE DATA SINK kin_dsink
LOCATION = 'kafka://kafka.abc.com:9092'
WITH OPTIONS
(
    CREDENTIAL = 'kafka_credential',
    KAFKA_TOPIC_NAME = 'kafka_topic'
)

Options

Option	Consumer	Description
`CREDENTIAL`	Any	Credential object to use to authenticate to the remote consumer
`VALIDATE`	Any	Whether to test the connection to the data sink upon creation; if `TRUE` (default), the creation of a data sink that cannot be connected to will fail; if `FALSE`, the data sink will be created regardless of connectivity
`WAIT TIMEOUT`	Any except `JDBC`	Timeout in seconds for reading from the consumer
`CONNECTION TIMEOUT`	Any except `JDBC`	Timeout in seconds for connecting to a consumer
`USE_HTTPS`	`AZURE`, `GCS`, `S3`	Whether to connect to the storage consumer over HTTPS or not `true`: Use HTTPS (default) `false`: Use HTTP
`USE_MANAGED_CREDENTIALS`	`AZURE`, `GCS`, `S3`	Whether to connect to the storage provider with consumer-managed credentials `AZURE`: Use the Azure Instance Metadata Service (IMDS) endpoint local to the Kinetica cluster to acquire credentials (only for on-prem clusters deployed within Azure) `GCS`: Satisfy the Application Default Credentials requirements for credentials, making sure any supporting server-side configuration (files, environment variables) is done on every node in the cluster (only for on-prem clusters deployed within GCS) `S3`: Use the AWS Default Credential Provider Chain to acquire credentials, making sure any supporting server-side configuration (files, environment variables) is done on every node in the cluster (only for on-prem clusters deployed within S3)
`KAFKA_TOPIC_NAME`	`KAFKA`	Kafka topic to write to
`JSON_FORMAT`	`HTTP`, `HTTPS`, `KAFKA`	The desired format of JSON encoded notifications message. `flat`: A single record is returned per message `nested`: Records are returned in an array per message
`MAX_BATCH_SIZE`	`HTTP`, `HTTPS`, `KAFKA`	Maximum number of records per notification message
`MAX_MESSAGE_SIZE`	`HTTP`, `HTTPS`, `KAFKA`	Maximum size in bytes of each notification message
`CONTAINER NAME`	`AZURE`	Azure storage container name
`SAS TOKEN`	`AZURE`	Azure storage account shared access signature token; this should be an account-level access token, not a container-level one
`STORAGE ACCOUNT NAME`	`AZURE`	Azure storage account name (only used if `TENANT ID` is specified)
`TENANT ID`	`AZURE`	Azure Active Directory tenant identifier
`GCS_BUCKET_NAME`	`GCS`	Name of the GCS bucket to use as the data sink
`GCS_PROJECT_ID`	`GCS`	Name of the Google Cloud project to use for request billing
`GCS_SERVICE_ACCOUNT_KEYS`	`GCS`	Text of the JSON key file containing the GCS private key
`DELEGATION TOKEN`	`HDFS`	Optional Kerberos delegation token for worker nodes; if not specified, the token will be acquired from HDFS
`KERBEROS KEYTAB`	`HDFS`	Location of the Kerberos keytab file in KiFS
`USE KERBEROS`	`HDFS`	Whether to attempt Kerberos authentication to HDFS
`JDBC_DRIVER_CLASS_NAME`	`JDBC`	JDBC driver class name (optional, if the name is available in the JAR file’s manifest)
`JDBC_DRIVER_JAR_PATH`	`JDBC`	KiFS path of the JDBC driver JAR file to use
`BUCKET NAME`	`S3`	Amazon S3 bucket name
`REGION`	`S3`	Amazon S3 region identifier
`S3_AWS_ROLE_ARN`	`S3`	Amazon Resource Name (ARN) specifying the role
`S3_ENCRYPTION_CUSTOMER_ALGORITHM`	`S3`	Algorithm used to encrypt/decrypt data
`S3_ENCRYPTION_CUSTOMER_KEY`	`S3`	Key used to encrypt/decrypt data
`S3_ENCRYPTION_TYPE`	`S3`	Server side encryption type
`S3_KMS_KEY_ID`	`S3`	KMS key
`S3_USE_VIRTUAL_ADDRESSING`	`S3`	Whether to use virtual addressing when referencing the S3 sink `true`: The requests URI should be specified in virtual-hosted-style format where the bucket name is part of the domain name in the URL `false`: Use path-style URI for requests
`S3_VERIFY_SSL`	`S3`	Whether to verify SSL connections `true`: Verify SSL connections (default) `false`: Don’t verify SSL connections; for testing purposes, bypassing TLS errors, self-signed certificates, etc.

Consumer-Specific Syntax

Several authentication schemes across multiple consumers are supported.

Azure

Syntax below, examples here.

CREATE DATA SINK [<data sink schema name>.]<data sink name>
LOCATION = 'AZURE[://<host>]'
WITH OPTIONS
(
	CREDENTIAL = '[<credential schema name>.]<credential name>',
	CONTAINER NAME = '<azure container name>'
)

CREATE DATA SINK [<data sink schema name>.]<data sink name>
LOCATION = 'AZURE[://<host>]'
WITH OPTIONS
(
    USE_MANAGED_CREDENTIALS = true,
    STORAGE ACCOUNT NAME = '<azure storage account name>',
    CONTAINER NAME = '<azure container name>',
    TENANT ID = '<ad tenant id>'
)

GCS

Syntax below, examples here.

CREATE DATA SINK [<data sink schema name>.]<data sink name>
LOCATION = 'GCS[://<host>]'
WITH OPTIONS
(
	CREDENTIAL = '[<credential schema name>.]<credential name>',
	[GCS_PROJECT_ID = '<gcs project id>',]
	GCS_BUCKET_NAME = '<gcs bucket name>'
)

CREATE DATA SINK [<data sink schema name>.]<data sink name>
LOCATION = 'GCS[://<host>]'
WITH OPTIONS
(
	USE_MANAGED_CREDENTIALS = true,
	[GCS_PROJECT_ID = '<gcs project id>',]
	GCS_BUCKET_NAME = '<gcs bucket name>'
)

CREATE DATA SINK [<data sink schema name>.]<data sink name>
LOCATION = 'GCS[://<host>]'
WITH OPTIONS
(
	[GCS_PROJECT_ID = '<gcs project id>',]
	GCS_BUCKET_NAME = '<gcs bucket name>'
)

CREATE DATA SINK [<data sink schema name>.]<data sink name>
LOCATION = 'GCS[://<host>]'
WITH OPTIONS
(
	GCS_SERVICE_ACCOUNT_KEYS = '<gcs account json key text>',
	[GCS_PROJECT_ID = '<gcs project id>',]
	GCS_BUCKET_NAME = '<gcs bucket name>'
)

HDFS

Syntax below, examples here.

Credential

CREATE DATA SINK [<data sink schema name>.]<data sink name>
LOCATION = 'HDFS://<host>:<port>'
WITH OPTIONS (CREDENTIAL = '[<credential schema name>.]<credential name>')

JDBC

Syntax below, examples here.

CREATE DATA SINK [<data sink schema name>.]<data sink name>
LOCATION = '<jdbc url>'
WITH OPTIONS
(
	[JDBC_DRIVER_CLASS_NAME = '<jdbc driver class full path>',]
	[JDBC_DRIVER_JAR_PATH = 'kifs://<jdbc driver jar path>',]
	CREDENTIAL = '[<credential schema name>.]<credential name>'
)

CREATE DATA SINK [<data sink schema name>.]<data sink name>
LOCATION = '<jdbc url with username/password>'

Kafka

Syntax below, examples here.

CREATE DATA SINK [<data sink schema name>.]<data sink name>
LOCATION = 'kafka://<host>[:<port>]'
WITH OPTIONS
(
	CREDENTIAL = '[<credential schema name>.]<credential name>',
	KAFKA_TOPIC_NAME = '<kafka topic name>'
)

CREATE DATA SINK [<data sink schema name>.]<data sink name>
LOCATION = 'kafka://<host>[:<port>]'
WITH OPTIONS (KAFKA_TOPIC_NAME = '<kafka topic name>')

Local (Kinetica)

Syntax below, examples here.

User Auth

CREATE DATA SINK [<data sink schema name>.]<data sink name>
LOCATION = 'TABLE://[<table schema name>.]<table name>'

S3

Syntax below, examples here.

CREATE DATA SINK [<data sink schema name>.]<data sink name>
LOCATION = 'S3[://<host>]'
WITH OPTIONS
(
    CREDENTIAL = '[<credential schema name>.]<credential name>',
    BUCKET NAME = '<aws bucket name>',
    REGION = '<aws region>'
)

CREATE DATA SINK [<data sink schema name>.]<data sink name>
LOCATION = 'S3[://<host>]'
WITH OPTIONS
(
    USE_MANAGED_CREDENTIALS = true,
    BUCKET NAME = '<aws bucket name>',
    REGION = '<aws region>'
)

CREATE DATA SINK [<data sink schema name>.]<data sink name>
LOCATION = 'S3[://<host>]'
WITH OPTIONS
(
    BUCKET NAME = '<aws bucket name>',
    REGION = '<aws region>'
)

Webhook

Syntax below, examples here.

CREATE DATA SINK [<data sink schema name>.]<data sink name>
LOCATION = 'https://<host>[:<port>]'
WITH OPTIONS (CREDENTIAL = '[<credential schema name>.]<credential name>')

CREATE DATA SINK [<data sink schema name>.]<data sink name>
LOCATION = 'http://<host>[:<port>]'

​Parameters

​Examples

​Options

​Consumer-Specific Syntax

​Azure

​GCS

​HDFS

​JDBC

​Kafka

​Local (Kinetica)

​S3

​Webhook

Parameters

Examples

Options

Consumer-Specific Syntax

Azure

GCS

HDFS

JDBC

Kafka

Local (Kinetica)

S3

Webhook