I have created a mapreduce program which reads some files from local and stores them to hdfs after processing. When I am trying to run this on HDP cluster it reads null value when all files are processed in the folder and throws NullPointerException. This problem is only coming in aws HDP cluster, whereas in our local dev environment all the components are installed separately and the same program is running perfectly. We are using hdp 2.6.2 version and all the components on dev environment are same as hdp 2.6.2. ... View more

I am trying to read zip file in nifi execute processor and I am using python as a scripting language. When I run the script it throws no viable alternative input at line 25 (flowFile = session.get()). What is the real cause behind this. Here is my script from zipfile import ZipFile from org.apache.nifi.processor.io import InputStreamCallback import java.io import json from org.apache.commons.io import IOUtils from java.nio.charset import StandardCharsets from org.apache.nifi.processor.io import StreamCallback class ReadVersion(InputStreamCallback): def __init__(self): self.flowFile = None self.version = '' self.error = '' def process(self,inputStream): try: zipname = self.flowFile.getAttribute('filename') zippath = self.flowFile.getAttribute('absolute.path') zfile = zipfile.ZipFile(zippath+zipname) with zipFile(zfile,'r') as zip: pageview = zip.read('pageview.json').decode("utf-8") pageview = json.loads(clients) pam = zip.read('pam.json') pam= json.loads(Company.decode("utf-8") ) flowFile = session.get() if (flowFile != None): callback = ReadVersion() callback.flowFile = flowFile session.read(flowFile, callback) if (callback.version != ''): flowFile = session.putAttribute(flowFile,'MSVersion',callback.version) session.transfer(flowFile, REL_SUCCESS) if (callback.error == 'error'): session.transfer(ff, REL_FAILURE) ... View more

I am building a job in which I have to validate phone numbers and we wants to use 'google-libphonenumber' npm package. I am using javascript in executescript processor. What is the correct way to include the npm package? ... View more

I am developing a JSON to CSV converter job in NIFI and I have to generate a UUID for each json element and add that generated UUID to the flowfile. I don't find any suitable processor for generating UUID ? My question is how can I generate a UUID for each incoming flowfile. ... View more

I want to convert Json coming from MongoDb to csv and I am using ConvertRecord processor for this purpose. I have configured all the required controllers and I am using AvroSchemaRegistry controller to validate the json and .JsonTreereader controller to read the json but when I run the job it it throws the error that schema not found. I have two question Which is the most proper way to do this Job. i.e Converting JSON to CSV ? How to pass schema into the JsonTreeReader This is ConvertRecord Processor Setting This my json file demo {"id":1001, "name":"vivek"} ... View more