Q&A

Hello, Greetings to the entire community, I have a problem, we currently have a nifi cluster with version 1.11.4, I need to send alerts to datadog, and the SiteToSiteStatusReportingTask processor, it helps me obtain error information in the flows that we have developed as I understand, however this processor does not appear for the version of Nifi that we have, is there any other way to obtain the information of the flows with errors or any maintenance of updating or putting a patch nifi-site-to-site-reporting-nar-1.11.4.nar I would appreciate any guidance or support.

Experts,Provide me jolt transform for my json input shown belowInput json would be like below. This is just one record. Actually there can be multiple records{"Product": "4039","Plant": "1010","MRPArea": "","PlndIndepRqmtType": "VSE","PlndIndepRqmtVersion": "00","RequirementPlan": "","RequirementSegment": "","PlndIndepRqmtIsActive": "X","PlndIndepRqmtPeriod": "20250502","NoWithdrawal": "","DeleteOld": "","PeriodType": "2","PlannedQuantity": "20.000","UnitOfMeasure": "PC","ProductionVersion": ""}Output what i am expecting as below{"Product": "4039","Plant": "1010","MRPArea": "","PlndIndepRqmtType": "VSE","PlndIndepRqmtVersion": "00","RequirementPlan": "","RequirementSegment": "","PlndIndepRqmtIsActive": "X","PlndIndepRqmtPeriod": "20250502","NoWithdrawal": "","DeleteOld": "","to_PlndIndepRqmtItem": [{"Product": "4039","Plant": "1010","MRPArea": "","PlndIndepRqmtType": "VSE","PlndIndepRqmtVersion": "00","RequirementPlan": "","RequirementSegment": "","PlndIndepRqmtPeriod": "20250502"

Hello, community. 😃I have an interesting case with NiFi (for me =P). We have:1. HDFS kerberized service that configured to use arcfour-hmac-md5 (rc4-hmac) for all default_tkt_enctype, default_tgs_enctypes and permitted_enctypes (yeap, it's legacy and now I cannot change it)2. Other kerberized applications, e.g. Kafka with disabled arcfour-hmac-md5 (allow_weak_crypto = false, permitted_enctypes excludes arcfour-hmac-md5)And I have no idea how to configure NiFi to communicate with that applications on time and use exactly supported TGS enctype for each one.If I set default_tgs_enctype in krb5.conf on NiFi to: default_tgs_enctypes = arcfour-hmac-md5 aes256-cts-hmac-sha1-96 aes128-cts-hmac-sha1-96 aes256-cts-hmac-sha384-192 HDFS with arcfour-hmac-md5 works fine, but Kafka (where arcfour-hmac-md5 is disabled) fails. And vise versa -- moving arcfour-hmac-md5 down of the default_tgs_enctypes list broke HDFS communications but Kafka starts to work.In krb5.conf I can overri

Hey,i'm using NIFI 1.22 and preparing NIFI flow for upgrade to 2.0.0 version.Currently, using  PutElasticsearchHttpRecord in order to index flowfiles which contain array of nested jsons. This processor sends each flowfile as whole, does not require splits to single jsons or defining schema.This processor is deprecated and it suggested to move to PutElasticsearchRecord.My data is very much complex and variant so I tried to avoid defining schema and to use  the Inffer Schema strategy of JsonTreeReader in  PutElasticsearchRecord.But the processor is failing on type conversion of the first field in the first json of the array. It seems that it cannot handle nested jsons in array (while splitting it to single jsons worked).In addition, my implemetation need to deal the following limitation: 1. Processing a large dataset (TBs), so it need to be fast.2. Avoiding spliting the array of jsons into a single json per flow file, as it cause content claim issues (too ma

Hey, communiy!We have a NiFi Cluster 1.18.0 (still so, yeap, sorry) and next issue do my mind.Simple flow, where we read data from Kafka with ConsumeKafka Processor and process it with EvaluateJsonPath after.One time we can see that queue between processors are stuck with data and nothing happens while EvaluateJsonPath or whole canvas not restared (right click -> stop -> start):Connection settings:Back Pressure Threshold: 10 000;Size Threshold: 1 GbLB Strategy: Round RobinQueue threshold exceeded only on 3rd cluster node: Additional configuration data:Maximum timer driven thread count: 400112 cores per node (2 CPU, 28 physical cores, 112 total with hyperthreading) So, as soon as I stop and start canvas -- all works fine again. Why it happens? How can I find the reason? 

Hello!I'm using kudu as my data store.But kudu leader master uses too much memory(about 30GB), and other masters and tablet servers uses much less(about 2GB). Leader master's memory usage Follower master's memory usage  Kudu Master spec:3 nodes16GB physical memory per master32GB swap memory per masterKudu Tablet spec:8 nodes16GB physical memory per tablet serverMetrics says 35335258112 byte(35.34 GB) are used by generic_heap_size.Does anyone knows about this? 

Hello,We have got a new requirement. Earlier the users were sending Soap request and our existing app was processing the soap request and then returning a soap response. We now want to do this using NiFi. We want users to be still sending the soap request and we will receive it in NiFi and then process it and return back a Soap response. Is there any way I can do this in NiFi?

I'm having a Hive table emp1 with 100 partitions in Text format.I want Spark to read emp table based on partitions bases and write to EMP2 in parquet format. How to achieve 1) 10 Partition Read from EMP and parallel write to emp2. Making sure multiple executors are running and processing the data in parallel to convert data in short period. I don't want to use HDFS path.

I am running into a problem with my DBCPConnectionPool service in Nifi and am hoping that someone has some experience with this. I'm using Nifi 2.0.0-M2.My current Database Connection URL isjdbc:postgresql://myurl:5432/mydatabase?currentSchema=public&stringtype=unspecifiedThis works great with no issues. However, I want to add another schema in there so I don't have to specify the new schema in my queries throughout all my nifi flows. For example, I was hoping this would work:jdbc:postgresql://myurl:5432/mydatabase?currentSchema=myschema,public&stringtype=unspecifiedBut when I do a query like "select * from mytable" with ExecuteSQL, I get an error: org.postgresql.util.PSQLException: ERROR: relation "mytable" does not exist If I change the query to "select * from myschema.mytable", it works.Chatgpt recommended I try this, but it also didn't work:jdbc:postgresql://myurl:5432/mydatabase?options=-c%20search_path=myschema,public&stringtype=unspeci

Hi,The repository https://repository.cloudera.com/artifactory/cdh-releases-rcs is not working and obtain the "Error 500 Internal Server Error"When I try to download the library https://repository.cloudera.com/artifactory/cdh-releases-rcs/org/apache/zookeeper/zookeeper/3.4.5-cdh5.4.0/zookeeper-3.4.5-cdh5.4.0.jar, I get a 500 error.On Friday the 12th, it was working.

Hello everyone,I need help in understanding if the NiFi Registry is appropriate for the kind of management I need to do in my caseI have a main nifi cluster, and the usual environments used in software development, but in the prod env the cluster is gonna be replaced by several different, indipendent clusters.I will also have a package logic, meaning that for example, one of the production clusters might have a certain processor group, where another one might not.Versioning will also managed through Git, does it conflict or cause any issues with the registry?Does the registry allow for updating only certain processor groups depending on the cluster i am working on, even if the shared dev env contains all the packages? Moreover, how does the registry work in the backend? For example, do ALL the PG get downloaded, even if only a few are gonna get installed? Security is the main concern here, because i can't download on a cluster a PG that is not supposed to be there, even if i am not gon

I'm working with Airflow jobs in CDE. In my case, I have a resource that contains two files: one is the DAG file, and the other is a custom Python script that I want to import into the DAG.According to the documentation, resources in Airflow are mounted under /app/mount/<PREFIX>. However, when a job is created or updated, the DAG is validated before the resources are actually mounted in that folder, which causes the import to fail.If I encapsulate the import inside a PythonOperator, it works without issues, because at that point the resources are already available and the file can be found. The error I get during validation is always something like:"failed: invalid request: Invalid DAG: dag_import: Traceback (most recent call last): File "<frozen importlib._bootstrap>", line 241, in _call_with_frames_removed File "/tmp/tmpfx_nrhlx/dag.py", line 6, in <module> from myModule import func ModuleNotFoundError: No module named 'myModule'"I’ve tried not only creating th

In a pyspark code which reads a hdfs path in it's respective format(text, orc, parquet) and writes it in parquet format in ozone path.Data is huge.1) How to do resource calculations for the pyspark jobLike no. of cores, no. of executors, memory allocation 2) Is there a way we can dynamically read the data from hdfs by adjusting according to it's file type. 3) What should be the optimal solution and approach for the data movement and what input mappings the approach should use.

Hi,We have a flow created with ListSFTP , we have it scheduled it once per day, but for some reason if files are not available at source, we would like to Stop it and execute Run Once and Start the Primary Processor again.These step work fine from NiFi UI but when we try to use NiFi Api (/processors/{id}/run-status) to stop the Primary processor, the Processor wont stop, we are passing the latest Version Id and Client Id as well.any suggestion on how to fix this ?

We have seen that a CDP AWS Datahub cluster was created but failed shortly after due to the master node becoming unreachable, resulting in errors like:1) Cluster creation timeout2) Unreachability3) Abrupt failure due to unhealthy nodesWe have seen that our custom service "MY_SERVICE" was ruled out based on cluster logs and suspect network or connectivity issues.We have also observed:The Cloudera Manager instance in our CDP AWS Datahub environment is currently marked as unhealthy and is not responding. AWS diagnostics show that the instance status check has failed at the OS level, indicating a deeper issue with the instance configuration itself.

Environment:- CDP Public Cloud (AWS)- DataHub cluster with CDH 7.3.1- Custom service: 'Service_Name'- Deployment via cluster template + pre/post recipesProblem:Cluster creation fails with the error:```No service type 'Service_Name' available for cluster with version 'CDH 7.3.1'```What we have done:- Created a valid 'Service_Name.jar' with:- 'descriptor/service.sdl'- 'cloudera-service.json' (includes '"cdhVersion": { "min": "7.3.1"}')- Used **pre-recipe** to download and place CSD and parcel files in '/opt/cloudera/csd/' and '/opt/cloudera/parcel-repo/'- Used **post-recipe** to distribute and activate parcels via Cloudera Manager API- Verified JAR contents and permissions- Still getting the same error during cluster creationERROR:Installation of CDP with Cloudera Manager has failed. Please find more details on Cloudera Manager UI. Failed command(s): ClusterTemplateImport(id=1546337329): java.lang.IllegalArgumentException: No service type 'Service_Name' available for cluster with version

How to run spark df.write inside UDF called in rdd.foreach or rdd.foreachpartitionI.e. spark session object inside executor.

I want to config haproxy as LB for nifi web. My nifi cluster is configured using self certificate. When accessing to nifi web, I got the error user unauthorization. Here is my config:global log 127.0.0.1 local2 chroot /var/lib/haproxy pidfile /var/run/haproxy.pid maxconn 4000 user haproxy group haproxy daemon stats socket /var/lib/haproxy/statsdefaults log global mode http option httplog option dontlognull timeout connect 5000 timeout client 50000 timeout server 50000 timeout http-request 10s timeout http-keep-alive 10frontend nifi bind *:80 mode http default_backend nifi http-request set-header X-ProxyScheme http http-request set-header X-ProxyHost %[hdr(host)] http-request set-header X-ProxyPort 80 http-request set-header X-Forwarded-For %[src] http-request set-header X-Forwarded-Proto httpbackend nifi mode http server nifi01 10.29.144.56:8443 ssl verify none server nifi02 10.29.144.57:8443 ssl

Hello,Based on this example in Iceberg doc I tried to create a table using a custom hadoop catalog.Using Hue in CDP 7.2 I performed SET iceberg.catalog.hadoop_cat.type=hadoop;SET iceberg.catalog.hadoop_cat.warehouse=s3a://my-bucket/hadoop_cat_warehouse;CREATE TABLE iceberg_hadoop_cat_table (id Integer,name String) STORED BY 'org.apache.iceberg.mr.hive.HiveIcebergStorageHandler' TBLPROPERTIES ('iceberg.catalog'='hadoop_cat'); But received Error while compiling statement: FAILED: Execution Error, return code 40000 from org.apache.hadoop.hive.ql.ddl.DDLTask. java.lang.IllegalArgumentException: Table location not set Did I miss something ? Is CDP suporting that ?Thanks,Laurent

 How can I ensure NiFi runs the correct Python script for each processor?Is there a proper way to structure and load multiple Python processors in NiFi?Any tip to debug or force NiFi to pick the updated script? 

In NiFi 2.4 and above, the Jython engine is no longer available in ExecuteScript.Previously, I used scripts like this to read and update attributes:val = flowFile.getAttribute('x')flowFile = session.putAttribute(flowFile, 'x', str(int(val) + 10))Now I want to do the same using Python in NiFi 2.4+:Read an attribute (e.g., x)Compute a new valueWrite it back as an attribute (or a new one)Keep the original FlowFile content unchangedWhat’s the recommended way in NiFi 2.4+?Any official docs, samples, or best practices for this scenario would be very helpful.

<userGroupProvider> <identifier>file-user-group-provider</identifier> <class>org.apache.nifi.authorization.FileUserGroupProvider</class> <property name="Users File">./conf/users.xml</property> <property name="Initial User Identity 1">nifi</property> <property name="Initial Group Identity 1"></property></userGroupProvider><accessPolicyProvider> <identifier>file-access-policy-provider</identifier> <class>org.apache.nifi.authorization.FileAccessPolicyProvider</class> <property name="User Group Provider">file-user-group-provider</property> <property name="Authorizations File">./conf/authorizations.xml</property> <property name="Initial Admin Identity">nifi</property> <property name="Initial Admin Group"></property> <property name="Node Identity 1"></p

We are running Apache NiFi 2.4 (NifiKop Helm chart) in a clustered environment.When NiFi attempts to start up and synchronize with Bitbucket Cloud Registry Client, the application fails to fully initialize.We have:72 process groups under version control (Bitbucket Cloud registry).nifi.web.jetty.threads=30 (max web threads).nifi-atlassian-nar-2.5.0-SNAPSHOT.narWe are observing inconsistent startup behavior:In some cases, NiFi starts successfully, but the logs contain repeated errors:java.io.IOException: too many concurrent streams (from the JDK HTTP/2 client used by the Bitbucket registry integration).Even though these errors appear, the cluster eventually loads, and the UI becomes available.In other cases, NiFi fails to start entirely. It gets stuck ( The UI is visible for a few minutes but then it appear like an empty canvas where you cannot change or create anything)  at the following log entry and does not progress further:2025-09-04 13:57:41,246 INFO [Framework Task Thread-27]

Hello All.What problems can there be when copying data between two clusters with different major versions if you use hdfs://... instead of webhdfs://...hadoop distcp hdfs://<namenode>:<port> hdfs://<namenode>Examle from documetntation -Copying between major versionsRun the distcp command on the cluster that runs the higher version of Cloudera, which should be the destination cluster. Use the following syntax:hadoop distcp webhdfs://<namenode>:<port> hdfs://<namenode>Note the webhdfs prefix for the remote cluster, which should be your source cluster. You must use webhdfs when the clusters run different major versions. When clusters run the same version, you can use the hdfs protocol for better performance.For example, the following command copies data from a Cloudera source cluster named example-source to another Cloudera version destination cluster named example-dest:hadoop distcp webhdfs://example-source.cloudera.com:8020 hdfs://example-dest.clouder

Dear Support Team,I hope this message finds you well.I am currently working on setting up Trino to query data from Hive. However, I am encountering difficulties related to Kerberos authentication.  Despite following available documentation, I have not been able to successfully connect to the Hive. I think there is something wrong with my KERBEROS settings. When I use telnet or ping, I can connect to HIVE correctly, but when I use trino CLI,  it shows that it is socket is closed by peer. I would appreciate your guidance on:1. I would like to ask whether I need to set up Kerberos separately for trino connection? 2. How can I get relevant information about kerberos authentication, such as keytab.Here are some details about my setup:（I think maybe at least this information is needed, but I don't know where to get them as following）1.hive.metastore.client.principal=trino.example.com 2.hive.metastore.client.keytab=trino.keytab .Please let me know if there are any files I shoul