Reading from Kinesis

<< Click to Display Table of Contents >>

Navigation:  Integrating SQLstream Blaze with Other Systems > Reading Data into s-Server > Reading from Other Sources  >

Reading from Kinesis

Previous pageReturn to chapter overviewNext page

The ECDA Kinesis agent lets you read data from Kinesis servers. The agent is started with code along the following lines. You can issue this code in SQLline or another JDBC client.

call sys_boot.mgmt.start_agent_hoster('myserver', 'kinesis', 'SCHEMA_NAME=TEST&TABLE_NAME=TEST&PARSER=CSV&CHARACTER_ENCODING=ISO-8859-1&SKIP_HEADER=false&INITIAL_POSITION_IN_STREAM=TRIM_HORIZON&REGION=us-west-1&KINESIS_STREAM_NAME=test&STREAMS_PER_KINESIS_STREAM=1&APPLICATION_NAME=testrun3&MAX_RECORDS_PER_GET_RECORDS=1500');

 

The first parameter ("myserver" above) is the name to be given to the agent. You can use this name to stop the agent.

Second parameter ("kinesis" above) is the agent type - in this case kinesis.

Third parameter is either the path to properties file or list of properties separated by '&'.

Stopping the Agent

To stop the agent, issue code along the following lines:

call sys_boot.mgmt.stop_agent_hoster('myserver');

 

Setting Up an AWS Profile Path

To read from Kinesis, you need to have an AWS configuration file set up. The xxx and yyy can be set up on the AWS Console as follows:

1.Open the AWS Console.

int_kinesis_aws_profile_path1

2.Click Identity & Access Management

int_kinesis_aws_profile_path2

3.Click Users.

int_kinesis_aws_profile_path3

4.Click your User ID.
5.Create an Access Key.

int_kinesis_aws_profile_path4

6.When you create an access key, the AWS console will allow you to download a credentials file which will contain the values for aws_access_key_id and secret_access_key.

Kinesis Management Functions

The predefined functions below help you manage reading from Kinesis.

SET_KINESIS_CLIENT

Creates a client kinesis client used in other management functions

create or replace procedure SET_KINESIS_CLIENT(

   name varchar(128), profilePath varchar(256), profileName varchar(128),

   endpoint varchar(256), regionId varchar(128)

)

language java

parameter style system defined java

no sql

external name 'sys_boot.mgmt.kinesis:com.sqlstream.aspen.namespace.kinesis.KinesisManagement.setKinesisClient';

 

LIST_KINESIS_STREAMS

Lists all kinesis streams for this client.

create or replace function LIST_KINESIS_STREAMS(

   clientName varchar(256))

returns TABLE (

      streamName varchar(256)

)

language java

parameter style system defined java

no sql

external name 'sys_boot.mgmt.kinesis:com.sqlstream.aspen.namespace.kinesis.KinesisManagement.listStreams';

 

CREATE_KINESIS_STREAM

Creates a kinesis stream for this client.

create or replace procedure CREATE_KINESIS_STREAM(clientName varchar(128), name varchar(256), numShards int)

language java

parameter style system defined java

no sql

external name 'sys_boot.mgmt.kinesis:com.sqlstream.aspen.namespace.kinesis.KinesisManagement.createStream';

 

DELETE_KINESIS_STREAM

Deletes a kinesis stream for this client.

create or replace procedure DELETE_KINESIS_STREAM(clientName varchar(128), name varchar(256))

language java

parameter style system defined java

no sql

external name 'sys_boot.mgmt.kinesis:com.sqlstream.aspen.namespace.kinesis.KinesisManagement.deleteStream';