Solved

Hello everyone, i need your help. I am relatively new to nifi and need your help. I am using the processor ConsumeKafkaRecord_2_6 1.15.3 with the record writer json set record writer. The output is then an array without object name. Which processor or setup can I use to add an array name to the json output? I get this Output  [ { "moving_id" : 17078440, "store_id" : 9, "gtin" : 4064813288877, "qty" : -1.0, "condition_indicator" : 40, "issue_date" : 1656585240000, "received_date" : 1656547200000, "update_date" : 1656578057133}, { "moving_id" : 17078441, "store_id" : 9, "gtin" : 4064813288907, "qty" : -1.0, "condition_indicator" : 40, "issue_date" : 1656585240000, "received_date" : 1656547200000, "update_date" : 1656578057133} ] and i need json file in this structur {"stocks":[ { "moving_id" : 17078440, "store_id" : 9, "gtin" : 4064813288877, "qty" : -1.0, "condition_indicator" : 40, "issue_date" : 1656585240000, "received_date" : 165654720

Hello Experts, I am trying to upgrade nifi from 1.12.1 version to 1.16.3.I have a script 'ECMAScript' which was working fine in 1.12.1and now behaving different in 1.16.3.Below is the portion of code giving error in 1.16 version when 'xyzId' is null.But it just logs null as expected in 1.12 version var xyzId = flowFile.getAttribute('xyzId');log.info(xyzId); Below is the error in 1.16 version when null value received for above line (log.info() line throws below error)imer-Driven Process Thread-8] o.a.nifi.processors.script.ExecuteScript ExecuteScript[id=bae38131-0181-1000-29af-196312cdd9d9] ExecuteScript[id=bae38131-0181-1000-29af-196312cdd9d9] failed to process due to org.apache.nifi.processor.exception.ProcessException: java.lang.RuntimeException: java.lang.NoSuchMethodException: Can't unambiguously select between fixed arity signatures [(org.apache.nifi.logging.LogMessage), (java.lang.String)] of the method org.apache.nifi.controller.TerminationAwareLogger.info for arg

Hi, I have successfully installed Cloudera Manager on RHEL 7 Azure VM but I'm not able to access the web ui. I have tried after adding 7180 port as inbound port in Network Security Group in azure portal.

Hi, I took the CDP generalist exam few weeks ago,and I can't find a place to download its invoice to hand it to my company for a refund.anyone can help? Regards,

The following attached error during Kerberos setup with Ambari 2.7.5  

Hi,I have some schedulers in OOZIE through hue and some of them some times fail. However when I run them manually after, they end successfully. Is there any way to put retry policy in my WFs? Here is the error that I am taking:Exit code of the Shell command 1<<< Invocation of Shell command completed <<<java.lang.reflect.InvocationTargetExceptionat sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)at java.lang.reflect.Method.invoke(Method.java:498)at org.apache.oozie.action.hadoop.LauncherAM.runActionMain(LauncherAM.java:410)at org.apache.oozie.action.hadoop.LauncherAM.access$300(LauncherAM.java:55)at org.apache.oozie.action.hadoop.LauncherAM$2.run(LauncherAM.java:223)at java.security.AccessController.doPrivileged(Native Method)at javax.security.auth.Subject.doAs(Subject.java:422)at org.apa

I am learning the ExecuteScript processor. I want to understand, how to use python script to fetch a file from local, similar to what the GetFile processor does.  Thanks

sql("show databases").show() sql("show tables").show()error 

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 attributebut non of the flowfiles ever go to success relationship queue.Here's my settings:What am I doing wrong?

I want to insert splitted records from splitjson processor in database table using nifi.For eg: Below is splitjson processor output. I want to push all the flow files in database table having same fragment.identifier.table columns are mentioned below:identity_id|json|person_refwhere identity_id should be auto sequence number json would  flow-file contentperson_ref is the foreign key having int data typeso  table will have 13 entries and identity_id column will be unique numbers in sequence, json column will have different flow file contents and person_ref column should be same for all 13 entries lets say 111.   

I have a pyspark dataframe with names likeN. PlainfieldNorth PlainfieldWest Home LandNEWYORKnewyorkSo. PlainfieldS. PlaindieldSome of them contain dots and spaces between initials and some do not. How can they be converted to: n Plainfieldnorth plainfieldwest homelandnewyorknewyorkso plainfields plainfield(with no dots and spaces between initials and 1 space between initials and name)I tried using the following but it only replaces dots and doesn't remove spaces between initials:names_modified = names.withColumn("name_clean", regexp_replace("name", r"\.",""))After removing the whitespaces and dots is there any way get the distinct values.like this.north plainfieldwest homelandnewyorkso plainfield

I'm trying to housekeep our hdfs filesystem and found alot of very old files that are owned by spark user. I don't remember creating a spark user that would generate these files, hence when I invoke an hdfs dfs -rm -R files...it's giving me a file ownership/permission error. Is there any work around I can bypass this? just like sudo in unix? Thank you. 

Hi,I have 3-node Nifi cluster.Please, i would like to ask if each node of the cluster can output different results (i.e no duplicates) from one another.SCENARIO:Let say i have 10 records on a DB.If node1 output the first record, i want node2 or node3 not to output first record again. Thank you.

Hi,   Does anyone know if Cloudera changed the URL to validate the Cloudera Certificate license number? I used to see the information from the URL 'https://certification.cloudera.com/verify/' and is not working now. Best,Sruthi Kumar

Hi,Please, kindly help.Use Case:Minifi should send logs to a Secured 3 Nodes Nifi Cluster; Method:I am trying to generate client certificate on one of the Nifi nodes(https://masternode:9443)to use for minifi configuration file.I ran this command:./tls-toolkit.sh client -c masternode -D 'CN=nifiadmin, OU=NIFI' -C /tmp/minifi/ -t nifinifinifinifi ERROR:tls-toolkit.sh: JAVA_HOME not set; results may vary[main] INFO org.apache.nifi.toolkit.tls.service.client.TlsCertificateAuthorityClient - Requesting new certificate from masternode:9443[main] INFO org.apache.nifi.toolkit.tls.service.client.TlsCertificateSigningRequestPerformer - Requesting certificate with dn CN=nifiadmin,OU=NIFI from masternode:9443Service client error: PKIX path building failed: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target Thank you.

When try to distribute Parcel file, I got errors like below.  << cloudera-scm-agent/cloudera-flood.log >>[27/Jun/2022 09:44:18 +0000] 23024 Thread-2 server INFO stats_alert CDH-6.3.4-1.cdh6.3.4.p0.6751098-el7.parcel: [1000] 0 0 0 0 120 0 0 80 0 0[27/Jun/2022 09:44:30 +0000] 23024 MainThread server INFO Adding torrent: file:///data/opt/cloudera/parcel-cache/CDH-6.3.4-1.cdh6.3.4.p0.6751098-el7.parcel.torrent[27/Jun/2022 09:44:30 +0000] 23024 MainThread rpc ERROR Failed handling RPC: AddTorrent Already present torrent: CDH-6.3.4-1.cdh6.3.4.p0.6751098-el7.parcelTraceback (most recent call last):File "/opt/cloudera/cm-agent/lib/python2.7/site-packages/flood/util/rpc.py", line 176, in executeRPCreturn fn(*args)File "/opt/cloudera/cm-agent/lib/python2.7/site-packages/flood/server.py", line 278, in addTorrenttorrent = self.depot.importTorrent(torrentUrl)File "/opt/cloudera/cm-agent/lib/python2.7/site-packages/flood/depot.py", line 101, in importTorrentraise TorrentError("

After configuring my Dockerfile with:ENV NIFI_WEB_HTTPS_HOST=0.0.0.0ENV NIFI_WEB_HTTPS_PORT=8443 I was able to run the container and access UI. If I increase replica, my cluster informations registries two address nodes 0.0.0.0:8443 and any changes I try to make I receive the error: Node 0.0.0.0:8443 is unable to fulfill this request due to: Transaction 00671bfc-20b8-4257-8590-8936e7265fde is already in progress. If I remove NIFI_WEB_HTTPS_HOST property, I cannot access UI, but log say that Jetty is running with container name, not 0.0.0.0. Is it possible to run nifi in a container without 0.0.0.0?

Hi,Please, I have a very large data to fetch incrementally( i need to store state) from an Oracle Database.And my query contains multiple JOIN statements.Please, what is the best processor to use for this scenerio because from the information i gathered i was told QueryDatabaseTableRecord proc cannot handle JOIN statements.Thanks.

I am using CM5.16 and using custom repositories but my issue is I am  not able to see the  Backup menu which is usually on the toolbar. Do i have to enable it?

I'm writing some pyspark code where I have a dataframe that I want to write to a hive table.I'm using a command like this. dataframe.write.mode("overwrite").saveAsTable(“bh_test”) Everything I've read online indicates that this should, by default, create  a managed table. However, no matter what I try, it always creates an external table. Is there a configuration setting somewhere that overrides the default behavior?

Please, kindly help.I am trying to do Minifi - Nifi s2s communication.So, i was trying to generate a flow file(to use for minifi) on one of Secured Nifi cluster nodes.- i dropped an input port on the root canvas & connected it to putfile processor- i created a process group. I dropped tailfile processor & Remote Process Group processor inside it.- I configured RPG with the url of the root canvasI don't actually know what i am doing wrong because all my configurations seems to be good.# Site to Site propertiesnifi.remote.input.host=masternodenifi.remote.input.secure=truenifi.remote.input.socket.port=10000nifi.remote.input.http.enabled=truenifi.remote.input.http.transaction.ttl=30 secnifi.remote.contents.cache.expiration=30 secs Below are the the pics. 

I believe I found a bug in the NiFi LookupRecord processor.When there is no match for a key in the DatabaseRecordLookupService it returns an error: The lookup processor config:The input JSON has the following structure: [ { "id": 1, "name": "Bitcoin", "symbol": "BTC" }, { "id": 2, "name": "Apecoin", "symbol": "Ape" }]   

Hi, moved Ni-Fi 1.16.3 folder from one destination to another and now getting error for ExecuteScript processor: 'Script Engine' validated against 'ECMAScript' is invalid because Given value not found in allowed set 'Clojure, Groovy, Lua, Python, ruby'. Any idea what's wrong? Thanks a lot. Ludvik

Sensor data coming from ConsumeMQTT processor looks like this{1,1,1,3465689,B4:E6:2D:34:E1:D9,BE:FB:E4:11:91:5A,-42,192.168.0.21,v0.16,15.88,16.35,46.59,95931.16,0,0,0,3.27,0,0,0,0}Expected Output: {    "val 1" : 1,    "val 2" : 1,.......}and so  on I'm preparing this incoming sensor data to be ingested in influxdb kindly help me out which processor is best for this and required configuration Thanks a lot

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 IOUtilsfrom java.nio.charset import StandardCharsetsfrom org.apache.nifi.processor.io import StreamCallbackimport 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 classflowfile = session.get()if(flowfile !=