使用NIFI中的PUTSQL获取存储过程的输出

Question

我必须在NIFI中获取Oracle存储过程的输出。 我已经尝试使用以下SQL语句的PUTSQL：

声明varchar2;开始pkg_test.p_test（1，out）;结束;

它工作正常，但只是执行脚本。 如何获得输出“输出”的价值？ 编辑：我在此处尝试了Groovy脚本：

https://community.cloudera.com/t5/support-questions/does-executesql-processor-plocessor-pallocessor-pallocessor-palle-to-execute-secute-stored-procedure/td-pdd-p/td-p/1589222

我会收到以下错误：

2022-06-17 13:38:53,353 ERROR [Timer-Driven Process Thread-9] oanpgroovyx.ExecuteGroovyScript ExecuteGroovyScript[id=26ab18f1-3b0c-18cf-d90b-3d5904676458] groovy.lang.MissingMethodException: No signature of method ：script6a6d0a35 $ _run_closure1.docall（）适用于参数类型：（字符串，字符串，java.sql.date，null，null，string，null，null，null，null，string ...）valut ，null，ok，null，null，...]：groovy.lang.missingmethodexception：无签名：script6a6d0a35 $ _run_closure1.docall（）适用于参数类型：（字符串，字符串，string，java，java.sql.date，null，null，null，null，null，null，null，null，null，null，null ，字符串，null，null，string ...）值：[xxxx，xxxx，2022-05-30，null，ok，ok，null，null，null，...]

at org.codehaus.groovy.runtime.metaclass.closuremetaclass.invokemethod（closuremetaclass.java：255）

，所以我有过程的输出，但我会收到错误！

脚本：

import org.apache.commons.io.IOUtils
import org.apache.nifi.controller.ControllerService
import org.apache.nifi.processor.io.StreamCallback
import java.nio.charset.*
import groovy.sql.OutParameter
import groovy.sql.Sql
import java.sql.ResultSet


////Get the session values from Nifi flow Start 
def flowFile = session.get()
if(!flowFile) return
  String TYPE_NOTIFICATION = flowFile.getAttribute('TYPE_NOTIFICATION')
  String ID_NOTIFICATION = flowFile.getAttribute('ID_NOTIFICATION')
////Get the session values from Nifi flow END
 
 

String sqlString ="""{call PKG_TEST.P_TEST(?,?,?,?,?,?,?,?,?,?,?)}""";

def parametersList = [ID_NOTIFICATION, TYPE_NOTIFICATION,Sql.VARCHAR,Sql.VARCHAR,Sql.DATE,Sql.VARCHAR,Sql.VARCHAR,Sql.VARCHAR,Sql.VARCHAR,Sql.VARCHAR,Sql.DATE ];
 
 
SQL.mydbxx.call(sqlString, parametersList) {out1, out2,...->
    flowFile.putAttribute("out1",out1)...
};

 

session.transfer(flowFile, REL_SUCCESS)

我存储过程的签名： 谢谢你！

Answer 1

我无法测试它 - 因此，它只是参考代码

executeGroovyScript处理器

add sql.mydb处理器级别上的参数并将其链接到必需的DBCP池。

将此大约设置为脚本主体

def ff=session.get()
if(!ff)return

def statement = '''
  declare out VARCHAR2; 
  begin
    PKG_TEST.P_TEST(?, out);
  end;
  ? := out;
'''
//parameters for each ? placeholder
def params = [
  ff.param_input as Long, //get parameter value from flowfile attribute
  SQL.mydb.VARCHAR, //out varchar parameter https://docs.groovy-lang.org/latest/html/api/groovy/sql/Sql.html#VARCHAR
]
SQL.mydb.call(statement, params){p_out-> //we have only one out patameter
    //closure to process output parameters
    ff.param_output = p_out //assign value into flowfile attribute
}

//transfer flowfile to success
REL_SUCCESS << ff