Friday, April 8, 2016

SQL in NiFi with ExecuteScript

There is a good amount of support for interacting with Relational Database Management systems (RDBMS) in Apache NiFi:

  • Database Connection Pool controller service: A shared resource for processors to get connections to an RDBMS
  • ExecuteSQL: A processor to execute SELECT queries against an RDBMS
  • PutSQL: A processor to execute statements (INSERT, UPDATE, e.g.) against an RDBMS
  • QueryDatabaseTable: A processor to perform incremental fetching from an RDBS table

I will have a blog soon describing the configuration and use of the QueryDatabaseTable processor, which was added in Apache NiFi 0.6.0.

To set up a Database Connection Pool controller service, refer to this User Guide section. I have configured mine for a local PostgreSQL instance:


NOTE: I named it 'PostgresConnectionPool', that will enter into the script config later.

Back to the title of this post :)  The scripting processors don't know about (or have a dependency on) the Database Connection Pool controller service instances or even the API (DBCPService interface, e.g.). This would often preclude our code from accessing the service to get a database connection.

However, DBCPService has a single method getConnection(), which returns a java.sql.Connection. This class is part of Java proper, and is all we really need from the service; we can talk JDBC from there.  One of the great things about Groovy is dynamic method invocation, meaning I can call a method on an object if I know the method is there, even if I don't know what Class is the type of the object.  We'll get to that shortly.

To work with a Controller Service from ExecuteScript, we need to get a reference to a ControllerServiceLookup from the process context. In Groovy, this looks like:
def lookup = context.controllerServiceLookup
Now that we have a lookup, we use it to locate the service we want. If we know the controller service we want (and it won't change, get deleted/recreated, etc.), we can get the identifier (Id below):


 However, for this example I wanted the user to be able to specify the name of the controller service (in this case PostgresConnectionPool), not the identifier. For that we need to get all controller service identifiers, then find the one whose name equals PostgresConnectionPool.

I used a dynamic property in the ExecuteScript config dialog to let the user set the name of the desired Database Connection Pool controller service:


ExecuteScript will create a variable for each dynamic property, and bind a PropertyValue object to it. This is so you can access the value of the property as the correct type (String, integer, etc.). In our case it's a String so we can use the PropertyValue.getValue() method. In Groovy it's as simple as:
def dbServiceName = databaseConnectionPoolName.value
Now that we have a ControllerServiceLookup and the name of the service we want to find, we can use the APIs to iterate over the services until we find one whose name is the one we're looking for:
def dbcpServiceId = lookup.getControllerServiceIdentifiers(ControllerService).find {
    cs -> lookup.getControllerServiceName(cs) == dbServiceName
}
Now we have the identifier of the service we want, and can use lookup.getControllerService(dbcpServiceId) to get a reference to the ControllerService itself.  Note we haven't referred to this service as a DBCPService, because the script (and the processor) do not have access to that class. However as I said before, we know the method (getConnection) we want to call and we have access to the return type of that method. So in Groovy you can just invoke it:
def conn = lookup.getControllerService(dbcpServiceId)?.getConnection()
I used the safe-reference operator (?.) but to go further you will want to ensure conn is not null, and report an error if it is.

Now that we have a Connection object, we could use the JDBC API (java.sql.*) to issue queries, go through the ResultSet's rows, then get each column's name and value, etc. etc. However Groovy has an excellent object called groovy.sql.Sql that does all this with Groovy idioms.  For example, to issue a query 'select * from users' and iterate over the rows (with their row number), you have:
def sql = new Sql(conn)
sql.rows('select * from users').eachWithIndex { row, idx ->
    // Do stuff for each row here
}
In my case, I want to find the column names the first time, and output them as CSV headers. Then for all rows I want to comma-separate the values and output:
if(idx == 0) { out.write(((row.keySet() as List).join(',') + "\n").getBytes()) }
out.write((row.values().join(',') + "\n").getBytes())
All that's left to do is to set the filename attribute to the one specified by the filename variable (see the Configure Processor dialog above), and transfer the new flow file.  The entire script looks like this:

import org.apache.nifi.controller.ControllerService
import groovy.sql.Sql

def lookup = context.controllerServiceLookup
def dbServiceName = databaseConnectionPoolName.value
def dbcpServiceId = lookup.getControllerServiceIdentifiers(ControllerService).find { 
    cs -> lookup.getControllerServiceName(cs) == dbServiceName
}
def conn = lookup.getControllerService(dbcpServiceId)?.getConnection()
try {
flowFile = session.create()
flowFile = session.write(flowFile, {out -> 
    def sql = new Sql(conn)
    sql.rows('select * from users').eachWithIndex { row, idx ->
        if(idx == 0) { out.write(((row.keySet() as List).join(',') + "\n").getBytes()) }
        out.write((row.values().join(',') + "\n").getBytes())
    }
  } as OutputStreamCallback)
  flowFile = session.putAttribute(flowFile, 'filename', filename.value)
  session.transfer(flowFile, REL_SUCCESS)
} catch(e) {
    log.error('Scripting error', e)
    session.transfer(flowFile, REL_FAILURE)
}
conn?.close()

This script probably needs a little work before you'd want to use it, to check whether the Controller Service was found, to quote any row value that has a comma in it, etc. But I tried to keep it brief to illustrate the concepts, which are the fluent NiFI API and the cool idioms of Groovy :) For my example table, the script produces a valid CSV file:


This processor is available as a template on Gist (here), as always I welcome all questions, comments, and suggestions.

Cheers!

19 comments:


  1. it’s ok to show some appreciation and say ‘great post’
    .NET developer

    ReplyDelete
  2. Hello,
    i have a question. I have a requirement to run a dynamic query in Hive. Can i achive that using executesql processor? i cannot use a SelectHiveQL processor because my query is dynamic. I tried it using executesql by creating a DBCP with Hive connection details but it is not working. Please help me.

    ReplyDelete
    Replies
    1. This comment has been removed by the author.

      Delete
    2. What do you mean by a dynamic query? Could you get the information into the NiFi flow and then use ReplaceText with NiFi Expression Language to specify the query?

      Delete
  3. Hey Matt, I am actually trying to transfer .WAF file from one DB to another DB, do you have any idea using what processor in NiFi is this possible?

    ReplyDelete
    Replies
    1. I'm not familiar with WAF files, can you explain more about them and which databases you are trying to transfer from/to?

      Delete
    2. This comment has been removed by the author.

      Delete
  4. Hi Matt:

    I am new to NiFi. Need to make a connection to PostgreSQL and load files into PostgreSQL using NiFi.

    Any help or examples greatly appreciated.

    Thanks.

    ReplyDelete
  5. Hey great post I am trying to do the same type of thing with HBase but get
    Caused by: groovy.lang.MissingMethodException: No signature of method: com.sun.proxy.$Proxy83.getConnection() is applicable for argument types: () values: []
    on this code
    import org.apache.nifi.controller.ControllerService
    flowFile = session.get()

    if(!flowFile) return

    def lookup = context.controllerServiceLookup
    def HbaseServiceName = HBaseConnectionName.value

    def HBaseServiceId = lookup.getControllerServiceIdentifiers(ControllerService).find {
    cs -> lookup.getControllerServiceName(cs) == HbaseServiceName
    }
    def conn = lookup.getControllerService(HBaseServiceId)?.getConnection()

    any suggestions? I am trying to implement increment method reusing HBase service.

    ReplyDelete
  6. Is this supportiv in NiFI 1.1.1 version, when tried it throws NPE ...

    ReplyDelete
    Replies
    1. It should be, do you know where the NPE is occurring?

      Delete
  7. can you help me how to create Database Connection Pool controller service

    ReplyDelete
    Replies
    1. Which part are you having trouble with? Have you seen the documentation for the controller service? https://nifi.apache.org/docs/nifi-docs/components/org.apache.nifi/nifi-dbcp-service-nar/1.2.0/org.apache.nifi.dbcp.DBCPConnectionPool/index.html

      Also, you may find more help from the Apache NiFi users list (just send a message to users-subscribe@nifi.apache.org

      Delete
    2. I follow all steps that are mentioned in link but now, I got issue "unable to execute SQL select * from tablename due to org.apache.nifi.processor.exception.ProcessExecption: org.postgresql.util.PSQLException: ERROR : relation "hcp" does not exist Position: 15; routing to failure"

      Delete
  8. try repeat AS-IS:

    ExecuteScript[id=015e1000-0b43-18ae-7b55-c488d71978a6] Scripting error: groovy.lang.GroovyRuntimeException: Ambiguous method overloading for method groovy.sql.Sql#.
    Cannot resolve which method to invoke for [null] due to overlapping prototypes between:
    [interface javax.sql.DataSource]
    [interface java.sql.Connection]

    ReplyDelete
    Replies
    1. Your "conn" variable is getting a null value assigned from the following code:

      def conn = lookup.getControllerService(dbcpServiceId)?.getConnection()

      I'm guessing the dbcpServiceId is not set or it is incorrect. You can remove the question mark from the above line to see if you get a NullPointerException when calling getConnection(). If so, then you did not find the Controller Service you were looking for

      Delete
  9. This comment has been removed by the author.

    ReplyDelete
  10. Hi Matt,
    Thanks for your post, I followed your instructions exactly but for some reason, flow works for few hours then I get connection problem.. it can't find the connectionPool

    can you help...

    this is the thread
    https://community.hortonworks.com/questions/144255/controller-service-is-disabled.html?childToView=144274

    ReplyDelete