org.thales.punch.libraries.storm.bolt

Class PostgresqlOutput

java.lang.Object

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

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

org.thales.punch.libraries.storm.bolt.PostgresqlOutput

All Implemented Interfaces:

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

public class PostgresqlOutput
extends AbstractJDBCOutput

This bolt will enable you to insert data into a Postgresql database.

REQUIREMENTS: The table on which you want to insert data should already be present inside a specific database. A preconfigured Postgresql 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 mytopology_table (text1 TEXT, text2 TEXT);

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 base on the table metadata on which you want to insert data.

The postgresql 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.

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

 
 dag: [
 {
   {
     component: generator
     type: generator_input
     settings:
     {
       messages:
       [
         {
           logs:
           {
             text1: this is a text 1
             text2: this is a text22
           }
         }
         {
           logs:
           {
             text1: this is a text 3
             text2: this is a text44
           }
         }
         {
           logs:
           {
             text1: this is a text 5
             text2: this is a text66
           }
         }
       ]
     }
    }
    {
      component: postgres
      type: postgresql_output,
      settings:
      {
        host: localhost,
        username: yuechun,
        port: 5432,
        password: yuechun,
        database: mytopology,
        table: mytopology_table,
        bulk_size: 1,
        pool_size: 10,
        pool_name: myPool,
        column_names:
        [
                text1
                text2
        ]
      },
      subscribe:
      [
        {
          component: generator,
          stream: log",
          fields:
          [
            text1
            text2
          ]
        }
      ]
    }
 ]
 

Author:

Jonathan YUE CHUN

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
PostgresqlOutput(org.thales.punch.libraries.storm.api.NodeSettings boltSettings)

Method Summary

Modifier and Type	Method and Description
protected Object	formatColumnForAutoCast(Object value)
void	prepare(Map stormConf, org.apache.storm.task.TopologyContext context, org.apache.storm.task.OutputCollector collector)

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

addTupleToQuery, cleanup, failAll, formatColumnForFunction, getComponentConfiguration, getDefaultColumnNameWrapperChar, getDefaultConvertBooleanToInt, process

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

PostgresqlOutput

public PostgresqlOutput(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

formatColumnForAutoCast

protected Object formatColumnForAutoCast(Object value)

Overrides:

formatColumnForAutoCast in class AbstractJDBCOutput