org.thales.punch.libraries.storm.bolt

Class OmniSciOutput

java.lang.Object

org.thales.punch.libraries.storm.api.BaseProcessingNode

org.thales.punch.libraries.storm.bolt.AbstractJDBCOutput

org.thales.punch.libraries.storm.bolt.OmniSciOutput

All Implemented Interfaces:

Serializable, org.apache.storm.task.IBolt, org.apache.storm.topology.IComponent, org.apache.storm.topology.IRichBolt

public class OmniSciOutput
extends AbstractJDBCOutput

This bolt will enable you to insert data into a MapD (OmniSci) database.

REQUIREMENTS: The destination table should already be present inside a specific database. A preconfigured OmniSci database on the system running this bolt. Data should already be valid data type representation (ex: if table contain Date column, your tuple should contain a field matching the column name and the data type representation in string). Each stream should contain unique field name.

create table flights(arr_timestamp timestamp, dep_timestamp timestamp, uniquecarrier varchar(50));

How it works:

Given the required parameters to connect to the database (user, password, database_name, port, host), this bolt will try to insert data while dynamically casting the inserted data to the right type, based on the destination table's metadata.

The OmniSci bolt will try to match the provided field name found in a tuple to the table column name on which you want to insert data. It will try to cast the data passed on each field to match your table schema dynamically.

The "jdbc" insert_type only allows single line insert. To insert multiple rows, use the "thrift_transport" insert_type.

Below is a valid configuration example you can use with the punchlinectl for testing purpose

 
 dag:
 [
   {
     component: generator
     type: generator_input
     settings:
     {
       messages:
       [
         {
           logs:
           {
             dep_timestamp: 2017-04-23 06:30:00
             uniquecarrier: test
             arr_timestamp: 2017-04-23 06:30:00
           }
         }
         {
           logs:
           {
             uniquecarrier: test
             arr_timestamp: 2017-04-23 06:30:00
             dep_timestamp: 2017-04-23 06:30:00
           }
         }
         {
           logs:
           {
             arr_timestamp: 2017-04-23 06:30:00
             dep_timestamp: 2017-04-23 06:30:00
             uniquecarrier: test
           }
         }
       ]
     }
   }
   {
     component: omnisci
     type: omnisci_output
     settings:
     {
       host: localhost
       port: 6274
       username: admin
       password: HyperInteractive
       database: omnisci
       table: flights
       insert_type: thrift_transport // can be jdbc (in that case, bulk_size should be set to 1)
       bulk_size: 3
       column_names:
       [
         arr_timestamp:cast
         dep_timestamp:cast
         uniquecarrier:cast
       ]
     }
     subscribe:
     [
       {
         component: generator
         stream: logs
         fields:
         [
           arr_timestamp
           dep_timestamp
           uniquecarrier
         ]
       }
     ]
   }
 ]
 

Author:

Jonathan YUE CHUN / Rafael LEVEQUE

See Also:

Serialized Form

Field Summary

Fields inherited from class org.thales.punch.libraries.storm.bolt.AbstractJDBCOutput

ackFail, bulkSize, COLUMN_NAME_WRAPPER_CHAR_SETTING, COLUMN_NAMES_SETTING, COLUMN_TYPE_CAST, COLUMN_TYPE_FUNCTION, columnNames, columnNameWrapper, columnsNamesSeparator, CONVERT_BOOLEAN_TO_INT_SETTING, convertBooleanToInt, database, expirationTimeout, hikariConfigMaps, hosts, jdbcHandleIterator, lastUpdateTimestamp, password, poolName, poolSize, queryQueueTime, reconnectionTimeout, SOURCE_TUPLE_FIELD_SETTING, sourceTupleField, table, username

Fields inherited from class org.thales.punch.libraries.storm.api.BaseProcessingNode

ackRate, collector, componentId, every, failRate, metricContext, nodeSettings, topologyContext, traversalTime

Constructor Summary

Constructors

Constructor and Description
OmniSciOutput(org.thales.punch.libraries.storm.api.NodeSettings boltSettings)

Method Summary

Modifier and Type	Method and Description
protected void	ackAll()
void	cleanup()
Map<String,Object>	getComponentConfiguration()
void	prepare(Map stormConf, org.apache.storm.task.TopologyContext context, org.apache.storm.task.OutputCollector collector)
void	process(org.apache.storm.tuple.Tuple input)

Methods inherited from class org.thales.punch.libraries.storm.bolt.AbstractJDBCOutput

addTupleToQuery, failAll, formatColumnForAutoCast, formatColumnForFunction, getDefaultColumnNameWrapperChar, getDefaultConvertBooleanToInt

Methods inherited from class org.thales.punch.libraries.storm.api.BaseProcessingNode

ack, declareOutputFields, enrichAndForwardMonitoringTuple, execute, fail, getNewPoint

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Detail

OmniSciOutput

public OmniSciOutput(org.thales.punch.libraries.storm.api.NodeSettings boltSettings)
              throws org.thales.punch.exceptions.ConfigurationException

Throws:

org.thales.punch.exceptions.ConfigurationException

Method Detail

prepare

public void prepare(Map stormConf,
                    org.apache.storm.task.TopologyContext context,
                    org.apache.storm.task.OutputCollector collector)

Specified by:

prepare in interface org.apache.storm.task.IBolt

Overrides:

prepare in class AbstractJDBCOutput

process

public void process(org.apache.storm.tuple.Tuple input)

Overrides:

process in class AbstractJDBCOutput

cleanup

public void cleanup()

Specified by:

cleanup in interface org.apache.storm.task.IBolt

Overrides:

cleanup in class AbstractJDBCOutput

getComponentConfiguration

public Map<String,Object> getComponentConfiguration()

Specified by:

getComponentConfiguration in interface org.apache.storm.topology.IComponent

Overrides:

getComponentConfiguration in class AbstractJDBCOutput

ackAll

protected void ackAll()