Member since
12-11-2019
8
Posts
2
Kudos Received
1
Solution
My Accepted Solutions
| Title | Views | Posted |
|---|---|---|
| 3087 | 02-05-2020 07:23 AM |
06-21-2022
07:02 AM
Hi.. You can find more information here
... View more
05-20-2022
11:20 AM
Hello everyone! In the company where I work they had the same problem, this time I had some time to review the script and do a proof of concept to try to solve the problem. In my test, after a few flowfiles the processor generates the oracle database connection error. Searching groovy documentation I found that the example that is floating around the internet is not correctly using the closure for the connection (this incident for example). The solution is to close the instance along with the connection, currently it is trying to close the connection that is declared at the beginning of the script but the instance is ignored. In my tests, when closing the connection instance, the process executed 30,000 flowfiles without generating any problems. Instead, for the definition of this script, after 14,000 flowfiles, the processor started to generate a connection error. You can find more info in the groovy documentancion: Connecting to HSQLDB Well, what was expected arrived... the solution is as follows: You need to add the finally statement and close sql //Imports
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 conncation string from NIFI conncation pool
def lookup = context.controllerServiceLookup
//Este valor se crea en el procesador con el nombre del controlador de la conexion
def dbServiceName = databaseConnectionPoolName.value
def dbcpServiceId = lookup.getControllerServiceIdentifiers(ControllerService).find {
cs -> lookup.getControllerServiceName(cs) == dbServiceName
}
def conn = lookup.getControllerService(dbcpServiceId)?.getConnection()
sql = Sql.newInstance(conn);
////Get the session values from Nifi flow Start
try {
flowFile = session.get()
if(!flowFile) return
PRIMER_PARAMETRO = flowFile.getAttribute('PRIMER_PARAMETRO')
SEGUNDO_PARAMETRO = flowFile.getAttribute('SEGUNDO_PARAMETRO')
// Procedimiento con 2 parametros de entrada y 2 de salida
String sqlString ="""{call PRUEBA_CONCEPTO.PROC_DUMMY_DUAL(?, ?, ?, ?)}""";
//Parametros de salida
def CODIGO
def MENSAJE
// Lista de parametros del procedimiento
def parametersList = [PRIMER_PARAMETRO, SEGUNDO_PARAMETRO, Sql.NUMERIC, Sql.VARCHAR];
//Ejecucion del procedimiento
sql.call(sqlString, parametersList) {out_status_code, out_status_desc ->
CODIGO = out_status_desc
MENSAJE = out_status_code
};
//Set the session values start
def attrMap = ['status_desc':CODIGO, 'status_code':String.valueOf(MENSAJE)]
flowFile = session.putAllAttributes(flowFile, attrMap)
session.transfer(flowFile, REL_SUCCESS)
} catch (e){
log.error('Scripting error', e)
flowFile = session.putAttribute(flowFile, "error", e.getMessage())
session.transfer(flowFile, REL_FAILURE)
} finally {
if (conn != null) conn.close();
if (sql != null) sql.close();
} In my case I use SQL.instance with a DBCPConnectionPool but it should be the same for this case. Solution Issue Dummy procedure: CREATE OR REPLACE
PACKAGE PRUEBA_CONCEPTO AS
/* TODO enter package declarations (types, exceptions, methods etc) here */
PROCEDURE PROC_DUMMY_DUAL ( PRIMER_PARAMETRO IN VARCHAR2
, SEGUNDO_PARAMETRO IN NUMBER
, CODIGO OUT NUMBER
, MENSAJE OUT VARCHAR2
);
END PRUEBA_CONCEPTO;
/
CREATE OR REPLACE
PACKAGE BODY PRUEBA_CONCEPTO AS
PROCEDURE PROC_DUMMY_DUAL ( PRIMER_PARAMETRO IN VARCHAR2
, SEGUNDO_PARAMETRO IN NUMBER
, CODIGO OUT NUMBER
, MENSAJE OUT VARCHAR2
) AS
BEGIN
-- TAREA: Se necesita implantación para PROCEDURE PRUEBA_CONCEPTO.PROC_DUMMY_DUAL
CODIGO := SEGUNDO_PARAMETRO;
MENSAJE := PRIMER_PARAMETRO;
END PROC_DUMMY_DUAL;
END PRUEBA_CONCEPTO; Test Procedure SET SERVEROUTPUT ON;
DECLARE
PRIMER_PARAMETRO VARCHAR2(100);
SEGUNDO_PARAMETRO NUMBER;
CODIGO NUMBER;
MENSAJE VARCHAR2(100);
BEGIN
PRIMER_PARAMETRO := 'eSTO ES UN MENSAJE';
SEGUNDO_PARAMETRO := 1;
PRUEBA_CONCEPTO.PROC_DUMMY_DUAL(PRIMER_PARAMETRO, SEGUNDO_PARAMETRO, CODIGO, MENSAJE);
DBMS_OUTPUT.PUT_LINE('CODIGO : ' || CODIGO);
DBMS_OUTPUT.PUT_LINE('MENSAJE : ' || MENSAJE);
END;
... View more
02-14-2022
04:44 AM
You need to use some processor to extract the info of the flowfile. For example, you can use ExtractText processor after generated table fetch proccesors. The result of this (ExtractText) processor is the atribute you need to use next (max and min) but depend how you configure you will have the same flowfile or not. This is important to the flow in the next steps (if you need to use again). You can find information of ExtractText processor here ExtractText processor And this is te documentation web page of nifi. Sorry for my english.
... View more
02-11-2022
04:42 AM
In my case, these parameters are filtered after a call to a web service, so my solution won't work for you. Can you show us how you're doing it?
... View more
02-05-2020
07:23 AM
1 Kudo
The nifi database client is different from the one used to develop the application. The specific problem was in a field of a source table of the view where this field was used both to record numerical values and text and on this field the to_number oracle function was used to convert the text to numeric. But since the configuration is different from the docker, the format of numbers was ',.' so the data could not be transformed. The data in the productive database were somewhat different from those in the development database. I used a function where i use TO_NUMBER(REPLACE(NVL(TRIM(p_cadena),'0'),'.',',')); to solved this problem. (Only for docker)
... View more
01-30-2020
07:37 AM
Hi all! I have a docker-compose created with an image of apache/nifi 1.10.0 into a linux machine. Something like this: version: '2'
services:
nifi-1:
image: 'apache/nifi:1.10.0'
hostname: nifi
ports:
- 8080:8080
- 8082:8082
environment:
- NIFI_WEB_HTTP_PORT=8080
- NIFI_ELECTION_MAX_WAIT=1 min
- TZ=America/Argentina/Buenos_Aires
- LANG=es_AR.UTF-8 Into my development i have a QueryDatabaseTable that read a entire oracle view and then insert this record in other oracle database table. My problem is that this in docker not run. If i run this in a local nifi server this run with success. QueryDatabaseTable configurations: The view: The problem (Im trying with a string field and a date field); The log 2020-01-30 12:26:14,004 ERROR [Timer-Driven Process Thread-10] o.a.n.p.standard.QueryDatabaseTable QueryDatabaseTable[id=72b4c9f0-e289-33d7-f82d-50bb8f23d4f1] Unable to execute SQL select query SELECT TRUNC(FECHA_EVALUACION) FECHA_EVALUACION,
CANT_ENCUESTADOS,
CANTIDAD_LIDERES,
COD_COMPETENCIA,
COD_EVALUACION,
COD_EVALUADO,
COD_EVALUADOR,
COD_MODULO,
COD_TIPO_EVALUACION,
DESC_COMPETENCIA,
DESC_EVALUACION,
DESC_MODULO,
DESC_TIPO_ENCUESTA,
DESC_TIPO_EVALUACION,
TO_CHAR(FECHA_EXAMEN,'DD/MONTH/YYYY') FECHA_EXAMEN,
NUM_FILA_EVALUADO,
NUM_FILA_MODULO,
NUM_FILA_TIP_EVALUACION,
GENERAL_NOTA_FINAL,
GENERAL_NOTA_POR_T_EVAL,
LIDERAZGO,
MODULO_CANTIDAD,
MODULO_NOTA,
MODULO_NOTA_FINAL,
MODULO_NOTA_PONDERADA,
MODULO_PONDERACION_TOTAL,
NOTA,
NRO_ENCUESTA,
PONDERACION_FINAL,
PONDERACION_FINAL_P_ENC,
PONDERACION_MODULO,
PONDERACION_MODULO_AJUSTADO,
NRO_ENCUESTA_REPETIDA,
TIPO_EVALUCION_CANT_P_ENC FROM VW_STG_CUB_EVALUACIONES due to java.sql.SQLSyntaxErrorException: ORA-01722: invalid number
: java.sql.SQLSyntaxErrorException: ORA-01722: invalid number
java.sql.SQLSyntaxErrorException: ORA-01722: invalid number
at oracle.jdbc.driver.T4CTTIoer11.processError(T4CTTIoer11.java:494)
at oracle.jdbc.driver.T4CTTIoer11.processError(T4CTTIoer11.java:446)
at oracle.jdbc.driver.T4C8Oall.processError(T4C8Oall.java:1054)
at oracle.jdbc.driver.T4CTTIfun.receive(T4CTTIfun.java:623)
at oracle.jdbc.driver.T4CTTIfun.doRPC(T4CTTIfun.java:252)
at oracle.jdbc.driver.T4C8Oall.doOALL(T4C8Oall.java:612)
at oracle.jdbc.driver.T4CStatement.doOall8(T4CStatement.java:213)
at oracle.jdbc.driver.T4CStatement.doOall8(T4CStatement.java:37)
at oracle.jdbc.driver.T4CStatement.executeForDescribe(T4CStatement.java:733)
at oracle.jdbc.driver.OracleStatement.executeMaybeDescribe(OracleStatement.java:904)
at oracle.jdbc.driver.OracleStatement.doExecuteWithTimeout(OracleStatement.java:1082)
at oracle.jdbc.driver.OracleStatement.executeQuery(OracleStatement.java:1276)
at oracle.jdbc.driver.OracleStatementWrapper.executeQuery(OracleStatementWrapper.java:366)
at org.apache.commons.dbcp2.DelegatingStatement.executeQuery(DelegatingStatement.java:329)
at org.apache.commons.dbcp2.DelegatingStatement.executeQuery(DelegatingStatement.java:329)
at org.apache.nifi.processors.standard.AbstractQueryDatabaseTable.onTrigger(AbstractQueryDatabaseTable.java:281)
at org.apache.nifi.controller.StandardProcessorNode.onTrigger(StandardProcessorNode.java:1176)
at org.apache.nifi.controller.tasks.ConnectableTask.invoke(ConnectableTask.java:213)
at org.apache.nifi.controller.scheduling.TimerDrivenSchedulingAgent$1.run(TimerDrivenSchedulingAgent.java:117)
at org.apache.nifi.engine.FlowEngine$2.run(FlowEngine.java:110)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Caused by: oracle.jdbc.OracleDatabaseException: ORA-01722: invalid number
at oracle.jdbc.driver.T4CTTIoer11.processError(T4CTTIoer11.java:498) So i don't know what happen. I have this development with other views and i don't have any problem but with this view is other thing. Can someone help me?
... View more
Labels:
- Labels:
-
Apache NiFi
-
Docker
12-11-2019
11:23 AM
1 Kudo
Hi all!! Im noob but i can execute a oracle procedure into a package using PutSql Processor. So enjoy it.. PutSql Processor configuration Oracle Sentence declare codigo NUMBER; mensaje VARCHAR2(500); begin pkg_name.prc_name(${block},${txseq},${item},codigo,mensaje); end; Package create or replace PACKAGE pkg_name is PROCEDURE prc_name ( p_block IN NUMBER, p_txseq IN NUMBER, p_item IN NUMBER, p_cod_error OUT NUMBER, p_msj_error OUT VARCHAR2 ); end pkg_name; Body create or replace PACKAGE BODY pkg_name IS PROCEDURE prc_name ( p_block IN NUMBER, p_txseq IN NUMBER, p_item IN NUMBER, p_cod_error OUT NUMBER, p_msj_error OUT VARCHAR2 ) IS v_id VARCHAR2(20); v_tag VARCHAR2(20); v_item_id VARCHAR2(20); e_error EXCEPTION; e_no_procesado EXCEPTION; BEGIN UPDATE paf_tablename paf SET paf.status = 'P', paf.cod_err = 0, paf.msg_err = 'Procesado' WHERE paf.block = p_block AND paf.item = p_item AND paf.txseq = p_txseq; p_cod_error := 0; p_msj_error := 'correcto'; EXCEPTION WHEN OTHERS THEN p_cod_error := 9000; p_msj_error := 'Error al procesar bloque ' || p_block || ' Transaccion ' || p_txseq || ' item ' || p_item || ' ' || substr(sqlerrm, 1, 100); END prc_name; END pkg_name; Sorry my english.
... View more