Hello! I've read   this   and   this   and   this   but explanations are pretty scarce. I understand what "&" means. But I can't wrap my head around what [&] means =\ Can someone please explain me what this [&] does applicable to this example? What exactly happening in lines? : "@1" : "outer[&4].inner[&2].t" , "@(3,eventTypeCode)" : "outer[&4].inner[&2].etc"   JSON: { "uID" : 1000358546 , "events" : [ { "eventTypeCode" : "FEEDBACK" , "transports" : [ "PUSH" , "SMS" ] } , { "eventTypeCode" : "MARKETING" , "transports" : [ "PUSH" , "EMAIL" ] } , { "eventTypeCode" : "ORDER_STATUS" , "transports" : [ "SOC_VK" ] } ] }   SPEC: [{ "operation" : "shift" , "spec" : { "events" : { "*" : { "transports" : { "*" : { "*" : { "@1" : "outer[&4].inner[&2].t" , "@(3,eventTypeCode)" : "outer[&4].inner[&2].etc" } } } } }, "*" : "&" } }] RESULT: { "uID" : 1000358546 , "outer" : [ { "inner" : [ { "t" : "PUSH" , "etc" : "FEEDBACK" } , { "t" : "SMS" , "etc" : "FEEDBACK" } ] } , { "inner" : [ { "t" : "PUSH" , "etc" : "MARKETING" } , { "t" : "EMAIL" , "etc" : "MARKETING" } ] } , { "inner" : [ { "t" : "SOC_VK" , "etc" : "ORDER_STATUS" } ] } ] }  Thanks beforehand! ... View more

Hello!   So I am trying to figure out how to work with the EnforceOrder processor. I have 200 flowfiles and every flowfile has attribute partition_number.  Many flowfiles has the same filename.   I wanna use EnforceOrder to merge them accroding to filename and partition_number attribute but non of the flowfiles ever go to success relationship queue. Here's my settings: What am I doing wrong? ... View more

Hello! I am trying to  read  JSON body from a flowfile and write the length of a JSON array of elements to new flowfile attribute. But somehow I am getting aforementioned error all the time. Here is my script.     from org.apache.commons.io import IOUtils from java.nio.charset import StandardCharsets from org.apache.nifi.processor.io import StreamCallback import json #Define a subclass of StreamCallback for use in session.write() class PyStreamCallback(StreamCallback): def __init__(self): pass def process(self, inputStream, outputStream): jsn = IOUtils.toString(inputStream, StandardCharsets.UTF_8) array = json.loads(jsn) # type: dict i = 0 while i <= 1: # ищем вложенность второго уровня, чтобы посчитать кол-во элементов (записей в таблице) root_key = list(array.keys())[0] array = array[root_key] i += 1 self.length = str(len(array)) def get_length_of_array(self): return self.length # end class flowfile = session.get() if(flowfile != None): flowfile = session.write(flowfile, PyStreamCallback()) flowfile = session.putAttribute(flowfile, "length", PyStreamCallback().get_length_of_array()) session.transfer(flowFile, REL_SUCCESS)     Can someone please explain me what am I doing wrong?   Thank you beforehand! ... View more

Hello!   So I can't access NiFi UI because the server if probably down exactly because of amount of files that have been stuck at queue. So the question is how can I find out the size of the queue for the moment using may be logs or maybe there are some other ways to do so?   Thank you beforehand! ... View more

So I am trying to implement this script     from datetime import datetime, timedelta, date flowfile = session.get() current_date= datetime.today().date() if flowfile != None: src_table_name = flowfile.getAttribute('src_table_name') date_filter = str(current_date.replace(day =11)) if src_table_name in ['Income_project_BP', 'Plan_final_scenario', 'Plan_primary_scenario', 'ProductionCalendar', 'Subcontracting_BP']: flowfile = session.putAttribute(flowfile, 'date_filter', date_filter) session.transfer(flowfile, REL_SUCCESS) else: session.transfer(flowfile, REL_FAILURE)       And I've tried a lot of different ways but when it comes to "replace" method I'm always getting this error.  Same thing when I am trying to use "strftime" method   What should I do? I really need this method. Thank you beforehand! ... View more

Hello! Sorry for my english. I am completely new to nifi and I am learning SplitText processor.   So here's the case. I have the comma separated txt file, something like this: KeyWord, SomeInformation   <---1st line is schema. KeyWord1, "information" KeyWord2, "information" KeyWord1, "another information" KeyWord2, "another information"   and so on.   So the question is how can I split this file on a few files based on KeyWord? So the every line with KeyWord1 go to one file, every line with KeyWord2 go to another file and so on?   Thank you beforehand! ... View more