Hello, I am using Nifi GetHbase Processor, and i want to get the number of records fetched by the GetHbase Processor or any how no of flowfiles it has generated. Is there any way to do this?Please help ... View more

Hello, I want to compile my java code using some external jars that are not included in the nifi lib folder.Is there any way to define a external path in the nifi, from where along with the nifi lib folder, my java code compiler will look into for the code compilation process? ... View more

Hello, I am going to make a template in which I use GetSftp processor, but I want to give path to fetch files dynamically which I get from database. For this I am using nifi expression language by which I set my dynamic value which comes from sql processor in attribute of a flow file. But now problem here is that GetSftp is a starting processor so it won't able to take input stream so I wont able to pass dynamic values to sftp. Please let me know some solution Thanks, ... View more

wait-notify-process-flow.xml Hello...above is my flow which i am currently running .I want to trigger my second flow after a certain condition is met in my first flow, which will start putting all the flowfiles to the wait processor and untill the second processor is completed and when it will complete, i want to resume my first flow.....But i am facing following issues ..... 1. I want to resume my first flow when my whole second flow is completed but notify processor receives a single flow file, it triggers the corresponding wait processor which is not suitable for me. 2. When my whole flow is completed like triggering second flow after flowfiles of first flow are waiting and then resumed execution after completion of my second flow , if i want to run the whole process again the wait processor of the first flow is not holding all the flowfiles in the queue like it did before, but processing it continuously , although my second flow has not even started , as it was supposed to be running after the completion of my second flow. Please find below the diagram of the flow. wait-notify-flow.png Below here the first flow wait-notify-first-flow-pic.png Below is the second flow wait-notify-second-flow-pic.png Here the second flow is in the left most part which is triggered by the wait processor in the left and the notify processor in the bottom will trigger the right most wait processor of the first flow. Here are the configurations of the processors distributed-map-cacheclientservice-config.png ... View more

Hello, I am using Nifi 1.3.0 with ExecuteSparkJob Processor and i am getting the following error in processor -- error.png and in the nifi-app.log --> 2017-07-14 21:53:44,231 INFO [stream error] c.t.nifi.v2.spark.ExecuteSparkJob ExecuteSparkJob[id=dd071c5c-c28e-3cc0-f9a5-ee5f965deaad] Diagnostics: File does not exist: hdfs://inno1.start.com:8020/user/root/.sparkStaging/application_1500027428726_0007/__spark_conf__8060965695322916343.zip Here is the configuration of the processor sparkjob.png. Please help me with the solution. ... View more

I have created a custom processor and after defining some properties it is showing " not a supported property" error as shown in the picture .Can anybody help me with this? ... View more

Hello...I am trying to convert my CSV file to get an AvroSchema by using my own custom processor which will get the file from the disk and converts it into flowfile and then passes it to the method named "inferAvroSchemaFromCSV" of the processor inferAvroSchema.After that i am getting my required schema successfully, which i am trying to transfer it into the success relationship so as to get on the disk, but i am getting following error - "DBColumnMapper[id=3836e9a0-015b-1000-9d9b-3d14d4066516] Failed to infer Avro schema for StandardFlowFileRecord[uuid=785128f9-acea-4d1c-a3dc-ec1313c900a0,claim=StandardContentClaim [resourceClaim=StandardResourceClaim[id=1491380551546-1, container=default, section=1], offset=15934, length=797],offset=0,name=vikas,size=797] due to org.apache.nifi.processor.exception.FlowFileHandlingException: StandardFlowFileRecord[uuid=785128f9-acea-4d1c-a3dc-ec1313c900a0,claim=StandardContentClaim [resourceClaim=StandardResourceClaim[id=1491380551546-1, container=default, section=1], offset=15934, length=797],offset=0,name=vikas,size=797] is not the most recent version of this FlowFile within this session (StandardProcessSession[id=12]): org.apache.nifi.processor.exception.FlowFileHandlingException: StandardFlowFileRecord[uuid=785128f9-acea-4d1c-a3dc-ec1313c900a0,claim=StandardContentClaim [resourceClaim=StandardResourceClaim[id=1491380551546-1, container=default, section=1], offset=15934, length=797],offset=0,name=vikas,size=797] is not the most recent version of this FlowFile within this session (StandardProcessSession[id=12]) 14:20:53 ISTERROR3836e9a0-015b-1000-9d9b-3d14d4066516" Following is my workflow template.png and my configuration of custom processor dbcolumnmapper-configuration.png Here is the Java Code-- @Override public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException { URI uri=null; String url=context.getProperty(INPUT_SCHEMA).getValue(); StringBuilder out = new StringBuilder(); try { uri = new URI(url); } catch (URISyntaxException e) { e.printStackTrace(); } Path schemaPath = new Path(uri); FileSystem fs; InputStream in=null; try { fs = schemaPath.getFileSystem(DefaultConfiguration.get()); in= fs.open(schemaPath); BufferedReader reader = new BufferedReader(new InputStreamReader(in)); String line; while ((line = reader.readLine()) != null) { out.append(line); } System.out.println("csv file data"+out.toString()); //Prints the string content read from input stream reader.close(); } catch (IOException e) { // TODO Auto-generated catch block e.printStackTrace(); } FlowFile flowFile=null; try{ //Writing csv data inside flow file for infer avro schema converter final byte[] dataByte=String.valueOf(out).getBytes(StandardCharsets.UTF_8); flowFile = session.create(); flowFile = session.write(flowFile, new OutputStreamCallback() { @Override public void process(final OutputStream output) throws IOException { output.write(dataByte); } }); //flowFile = session.putAttribute(flowFile, CoreAttributes.MIME_TYPE.key(), "text/plain"); //flowFile = session.putAttribute(flowFile, CoreAttributes.FILENAME.key(), "vikas"); final AtomicReference<String> avroSchema = new AtomicReference<>(); avroSchema.set(inferAvroSchemaFromCSV(flowFile, context, session)); FlowFile avroSchemaFF = null; avroSchemaFF = session.write(session.create(), new OutputStreamCallback() { @Override public void process(OutputStream out) throws IOException { out.write(avroSchema.get().getBytes()); } }); avroSchemaFF = session.putAttribute(avroSchemaFF, CoreAttributes.MIME_TYPE.key(), AVRO_MIME_TYPE); avroSchemaFF = session.putAttribute(avroSchemaFF, CoreAttributes.FILENAME.key(), "newFlowFile"); session.transfer(avroSchemaFF, REL_ORIGINAL); session.transfer(avroSchemaFF, SUCCESS); //session.transfer(avroSchemaFF, FAILURE); session.commit(); }catch (Exception ex) { getLogger().error("Failed to infer Avro schema for {} due to {}", new Object[] {flowFile, ex}); } } ... View more

I have a data-file which have say 10 records with primary keys 1-10 and i have some data in my database,say 5 records with primary keys 1-5. I want to insert that file data into my database which will update my 5 records which is having same primary key and insert the remaining 5 records, a total of 10 records in database.Here is my workflow insertupdateworkflow.xml , insertupdateworkflow.png,in this , i am first trying to update the records and redirecting the failed records to another jsontosql processor with insert statement so as to insert the remaining five records in the database.But the putsql is giving error "Index array out of bounds" and not inserting the data but is updating the existing data correctly.Please help me out with this. ... View more