NiFi Connector Developer Manual

• <https://github.com/kineticadb/kinetica-connector-nifi>

​

Building the Kinetica NiFi Connector

1. Download the connector source:
   $ git clone https://github.com/kineticadb/kinetica-connector-nifi.git
   $ cd kinetica-connector-nifi
   
2. If using a version of NiFi other than 1.3.0, update the pom.xml file with the correct version of NiFi in this block:
   <parent>
       <groupId>org.apache.nifi</groupId>
       <artifactId>nifi-nar-bundles</artifactId>
       <version>1.3.0</version>
   </parent>
   
3. Build the connector jar:
   $ mvn clean package
   

​

Installing the Kinetica NiFi Connector into NiFi

$ cp nifi-GPUdbNiFi-nar/target/nifi-GPUdbNiFi-nar-1.3.0.nar <NiFiHome>/lib

​

Getting Streaming Data from Kinetica to JSON or CSV Files

1. Drag a new Processor onto the flow
   • Select the GetKineticaToJSON or GetKineticaToCSV type
2. Properties tab
   • Server URL: The URL of the Kinetica instance you are using. This will be:
     • Format: http://<db.host>:9191
     • Example: http://localhost:9191;CombinePrepareAndExecute=1;RowsPerFetch=20000
   • Table Name: The name of the table to read from
   • Table Monitor URL: The URL Kinetica will be using to forward any new data inserted into the above table. This will be:
     • Format: tcp://<db.host>:9002/
     • Example: tcp://localhost:9002
   • Delimiter: For CSVs, the delimiter used in the file (e.g., comma, tab, pipe, etc.); defaults to tab
   • Username: Kinetica login username
   • Password: Kinetica login password

​

Saving Data to Kinetica Using NiFi Attributes

1. Drag a new Processor onto the flow:
   • Select the PutKinetica type
2. Settings tab:
   • Under Auto terminate Relationships, check the failure and success options.
3. Properties tab:
   • Server URL: The URL of the Kinetica instance you are using. This will be:
     • Format: http://<db.host>:9191
     • Example: http://localhost:9191;CombinePrepareAndExecute=1;RowsPerFetch=20000
   • Collection Name: Set this value if you want the table created in a collection.
   • Table Name: The name of the table to write to
   • Schema: A CSV string, where each entry is:
     • Format: <fieldname>|<data type>[|<subtype>]
     • Example:
       X|Float|data,Y|Float|data,TIMESTAMP|Long|data,TEXT|String|text_search
       
     For more details on schemas, read the Kinetica documentation.
   • Batch Size: The size of the batch to compress for efficient loading
   • Username: Kinetica login username
   • Password: Kinetica login password
   • Update on Existing PK: If a primary key (PK) is defined for a table, then there are two options for handling each new record pending insert that has a PK value matching an existing record in the target table. If set to true, the record in the target table will be updated with the new record’s values; if false, the new record will be discarded; defaults to false
   • Replicate Table: If true, the target table will be replicated; if false, the table will be distributed; defaults to false
   • Date Format: The date format to use to parse values in any datetime fields (e.g., dd-MM-yyyy hh:mm:ss)
   • TimeZone: Provide the timezone if the date is not from your local timezone
4. Specifying data to be saved into Kinetica:
   • Place processors upstream from this which assigns values to user-defined attributes named <field name>, where <field name> is the name of a field in your table
   • Each record written to your table will contain field values of:
     • the value in the attributes with names <field name> or
     • the value of null if no attribute is found with that field name

​

Saving Data to Kinetica Using Delimited Files

1. Drag a new Processor onto the flow
   • Select the PutKineticaFromFile type
2. Settings tab:
   • Under Auto terminate Relationships, check the failure and success options.
3. Properties tab:
   • Server URL: The URL of the Kinetica instance you are using. This will be:
     • Format: http://<db.host>:9191
     • Example: http://localhost:9191;CombinePrepareAndExecute=1;RowsPerFetch=20000
   • Collection Name: Set this value if you want the table created in a collection.
   • Table Name: The name of the table to write to
   • Schema: A CSV string, where each entry is:
     • Format: <fieldname>|<data type>[|<subtype>]
     • Example:
       X|Float|data,Y|Float|data,TIMESTAMP|Long|data,TEXT|String|text_search
       
     For more details on schemas, read the Kinetica documentation.
   • Delimiter: The delimiter used in the file (e.g., comma, tab, pipe, etc.); defaults to ,
   • Escape Character: The character used to escape other characters in the data (e.g., \); defaults to "
   • Quote Character: The character used to quote column data in the file (e.g., " or '); defaults to "
   • File Has Header: Whether the first line of the file is a header row or not; defaults to true
   • Batch Size: The size of the batch to compress for efficient loading
   • Error Handling: If true, the processor will skip rows that can’t be loaded successfully (due to parse error, etc.); if false, the processor will stop loading as soon as an error occurs; defaults to true
   • Username: Kinetica login username
   • Password: Kinetica login password
   • Update on Existing PK: If a primary key (PK) is defined for a table, then there are two options for handling each new record pending insert that has a PK value matching an existing record in the target table. If set to true, the record in the target table will be updated with the new record’s values; if false, the new record will be discarded; defaults to false
   • Replicate Table: If true, the target table will be replicated; if false, the table will be distributed; defaults to false
   • Date Format: The date format to use to parse values in any datetime fields (e.g., dd-MM-yyyy hh:mm:ss)
   • TimeZone: Provide the timezone if the date is not from your local timezone
4. Create a connector between the data source processor and the PutKineticaFromFile processor
   • Details tab: check the with coordinates option.