Q&A

Hi All, I am using putdatabaserecord to store a value into DB.Now i have a flow based on failure link to (this processor), that this failed record will be stored in disc for troubleshooting.I can store the failed record in disc by using putfile processor.Now i wanted it more informative during debugging, as the attribute "putdatabaserecord.error" contains the error reason, i want to append the flowfile content with this error reason.  I tried to acheive this using below flow..putsql/putdbrecord -> failure-> updateattribute (this is to give the flowfile a realistic name)-> attributesToJson (destination flowfileattribute/content -> putfilebut in attributestoJson , if the destination is kept as flowfileattribute, the flowfile dont get this attribute or if it is kept as flowfile content, original flowfile content is replaced...My requirement is to append this single attribute to the existing flowfile content. Any help or suggestion is greatly appreciated.

I have an InvokeHTTP processor Apache NiFi, I need to send a curl request with this contentcurl https://reqbin.com/echo/post/json-X POST-H 'Content-Type: application/json'-d '{ "login": "my_login", "password":"my_password"}'I can't figure out what to do -d

Hi Community,hope someone can help me to build this transform: My Input:{"Orders": [{"Headers": {"UniqShipment": "0"},"Goods": [{"GoodsTypeID": 3,"GoodsTypeName": "ALTRI MOBILI LEGNO","GoodsDetails": [{"Packs": 1}, {"Packs": 1}, {"Packs": 1}]}],"References": [{"TypeReference": "DT","ValueReference": "006611","DateReference": "2023-04-14"}, {"TypeReference": "CM","ValueReference": "JOHN"}, {"TypeReference": "OC","ValueReference": "V1250-0010"}, {"TypeReference": "RG","ValueReference": "TRCA_CATR726240_CA_006611_14_04_2023.json"}]}]}My Desired Output:{ "header": { "UniqShipment": "0" }, "rows": { "GoodsTypeName": ["ALTRI MOBILI LEGNO", "ALTRI MOBILI LEGNO", "ALTRI MOBILI LEGNO"], "Packs": [1, 1, 1], "ValueReference": ["V1250-0010", "V1250-0010", "V1250-0010"] }}My Transformation so far:[ { "operation": "shift", "spec": { "Orders": { "*": { "Headers": "header", "Goods": { "*": { "GoodsDetails": { "*": {

While trying to connect to a Kafka feed suddenly the connection drops out and the socket gets disconnected.Error :Caused by: org.apache.kafka.common.errors.SaslAuthenticationException: Failed to configure SaslClientAuthenticatorCaused by: org.apache.kafka.common.errors.SaslAuthenticationException: Failed to create SaslClient with mechanism GSSAPI[WARN ] 2023-05-20 11:52:25 [main] NetworkClient - [Consumer clientId=consumer-R_Consumer-1, groupId=R_Consumer] Bootstrap broker 10.49.148.50:9093 (id: -4 rack: null) disconnected[WARN ] 2023-05-20 11:52:25 [main] NetworkClient - [Consumer clientId=consumer-R_Consumer-1, groupId=R_Consumer] Error connecting to node 10.49.148.51:9093 (id: -3 rack: null)java.io.IOException: Channel could not be created for socket java.nio.channels.SocketChannel[closed]

When parsing a json array with a json tree reader (in this case using convert record), I am presented with this error, this testing was done in Nifi Version 1.21.0. However this error never occurs in Version 1.18.0. An example of json that fails in 1.21.0 that works in 1.18.0 [ {"a": [ {"b": [{}]}]}, {"a": [ {"b": []}]}]   Any help would be greatly appreciated.

Does Apache NiFi supports MQTT 5? If it does, how do i get the Response Topic?

I have a deployed model un Cloudera Machines Learning. The model is trained once a month, which I do through a pipeline (Jobs). It must be deployed after that again to take the updated model. Can the deployment be done in a way other than manual? 

I would like to get records count(select count(1) from table where col1=‘abc’) from a sql table. Could someone please help me with that ? 

I am not able to change my company name in my profile, as below shows.I also noticed that  the system is now showing that "Partner (No Longer Active)".not sure how to handle above two issue. 

There is a Quorum Kafka cluster with parameters.  listeners=PLAINTEXT://:9092,CONTROLLER://:9093, SASL_PLAINTEXT://:9094security.inter.broker.protocol= SASL_PLAINTEXTsasl.mechanism.inter.broker.protocol=PLAINsasl.enabled.mechanisms=PLAINallow.everyone.if.no.acl.found=truesuper.users=User:adminadvertised.listeners=SASL_PLAINTEXT://:9092, PLAINTEXT://:9093  Here you can see that port 9094 is set to SASL_PLAINTEXT.In nifi, I can write to a topic using the PublishKafka_2_6 processor, using port 9094.But it is impossible to read from the same topic and port 9094. The processor does not show any errors. I used different versions of the consumekafka processor. No results.I am able to read from the topic on the insecure port 9092. I am also able to read on the SASL_PLAINTEXT port through other tools such as datagrip. But for some reason nifi does not want to read through this port. help me please.nifi version 1.16.3

Hey there,I need some help with configuring EvaluateJsonpath because I'm having trouble with it. Everything else seems to be working fine, but I always get stuck at this part. I've tried looking for solutions online, but none of them seem to work or they're just outdated. It's possible that I messed up something in the other function blocks, but they still work if I remove the EvaluateJsonpath. Any ideas?  This is the json file:{ "random": 3171504720, "als": 29024, "mic": 47334, "gyro": { "x": -26, "y": 8, "z": 1 }, "accel": { "x": -1829, "y": 447, "z": 16802 }} 

Hi All, I have Nifi running on EKS cluser with 3 nodes. Every now and then, flow.xml.gz gets corrupted in one of the node and it fails to start back up again. Same things happens with other nodes sooner or later and ultimately NiFi crashes. The error from nodes startup logs - schema validation Error parsing Flow configuration at line <line_no>, col <col no > : cvc-complex-type:2.4.d : Invalid content was found starting with element 'property'. No child element is expected at this point.It has become a pain to maintain Nifi because of this issue. Please suggest a solution.

HelloI have the following :Tomcat server (local)Nifi (local)customer csv fileIn my flow I am getting the file content and transforming it to a json body (the body I need to send it in the API post).The problem is to use the invokeHTTP processor to send the api (onboard) I need to get a token via another API (auth) and add it in the header of the main api (onboard). Auth API : Sends a json to the tomcat :  { "user_name": "User", "password": "Password"}  And get a json with a token valid for 2 min  { "id": 1, "user_name": "user", "token": "eyJhbGciOiJIUzUxMiJ9.eyJzdWIiOiJzdXBlcmFkbWluIiwidGV...", "roles": [ "USER", "ADMIN" ], "homePage": null, "priorite": 0}   The problem is when I pass the customers json to the auth API it will take that json as the body for that API and I lose my customer json.Is there another way to get the token from the auth API and use it in the onboard API header withou

We have several CDSW jobs set up where we call a python script with a main() function which is called via if __name__ ==  "__main__":    main() Our problem is that these jobs are erronously shown as "Success" even if the python code throws an Exception. How can we set up our CDSW such that these jobs are correctly shown as "Failure"?

I am using Nifi parameter context to avoid hardcoding multiple variables.I used toolkit to create parameter context and used params inside my flow.Now I want to update some values of this existing parameter context.. 1. to update some values or 2. to add new parameters.I could not find a way using toolkit in the documentation. I see an answer here https://community.cloudera.com/t5/Support-Questions/Update-parameter-contexts-dynamically-using-REST-API/m-p/347697 But i am looking for any solution based on toolkit or other easier approach, so that it can be easily handled while deployments to next environments or while updating it in production environments. Please help me with your approach or suggestions.

I am trying to update parameter context within Nifi which is allocated to several process groups. We would like to update the value of one parameter within the parameter context. Please provide some sample payload to do to update parameter contexts either PUT or POST method

Hi, I am using hive 3.1.2 with hadoop 3.3.5. I manually trigger minor compaction but it's stuck in initiated state. I can't find any documentation on how to configure for workers. I read somewhere it is automatical when you set hive.txn.manager to org.apache.hadoop.hive.ql.lockmgr.DbTxnManager and hive.support.concurrency to true. I have also set below parameters but it is still stuck in initiated state. $ hive -e 'show compactions' 2>/dev/nullCompactionId Database Table Partition Type State Worker Start Time Duration(ms) HadoopJobId1 default employee_trans --- MINOR initiated --- --- --- --- hive.compactor.abortedtxn.threshold=200hive.compactor.check.interval=100shive.compactor.cleaner.run.interval=500mshive.compactor.compact.insert.only=truehive.compactor.delta.num.threshold=2hive.compactor.delta.pct.threshold=0.1hive.compactor.history.reaper.interval=2mhive.compactor.history.retention.attempted=2hive.compactor.history.retention.failed=3hive.compactor.hist

I want to connect opensearch with nifi but it is showing error while running processor of sending data. it shows failure of requests . i have also tried different version of nifi with opensearch but still same error . How can i solve it? screenshot of error 

I followed this guide on the Cloudera website, I've finished "Installing and Upgrading the Sentry Service" step. Now what do I do next?I tried to start Beeline and execute some queries as normal but get a privilege errorkinit -k -t /home/vgdata/vega.keytab vega@BI.VEGA.COM  (get tgt for user vega)beeline -u "jdbc:hive2://data-node01:10000/test;principal=hive/data-node01.vega.com@BI.VEGA.COM" (access beeline with kerberos principal)select * from test; (execute query) But I got an error like belowError: Error while compiling statement: FAILED: SemanticException No valid privilegesUser vega does not have privileges for SWITCHDATABASEThe required privileges: Server=server1->Db=*->Table=+->Column=*...User "vega" is hdfs superuser, and also the main user that we use to connect to all Hadoop services (there is a Kerberos principal "vega" too)I tried to execute some commands likecreate role admin;grant role ....But all get an error like "No groups found for user ve

Hi All,I would like to know what is heap value that I need to set for SNN ?I have a big cluster with ~600 nodes and ~150M blocks so I had set NN heap to be 150GB but what value should I keep for SNN ?Currently its running with 8GB, I feel its low as it needs more MEM to perform checkpointing. Each hosts are running with 750GB of MEM.Please suggest.

I'm fascinated by Apache Airflow, but I can't operate it locally. Is there anywhere I can test it on the cloud? Almost all of the other posts in this sub are about Airflow/Dagster.For example, we have replit to experiment with Python and other programming languages; is there a comparable infrastructure for Airflow/Dagster? Thank You.

When I make some changes to some service configurations, Cloudera asks me to "restart stale service". And there is a window pop-up asking me should I "Redeploy Client Configuration" too like below:The default option is No (not tick), so does this mean this is optional? When should I tick this? Since if I tick this, it will take a long time to restart everything, even I only make a small change. I wonder if is it necessary.

             i'm trying to split the array values to seperate list of object values but not able to get it.input:{"FatturaElettronicaBody": {"DatiGenerali": {"DatiGeneraliDocumento": {"TipoDocumento": "TD01","Numero": "126561","Data": "05.05.2023","Divisa": "EUR","ImportoTotaleDocumento": "9382.29"},"DatiDDT": {"NumeroDDT": "086708,086704,086705,086706,086707"}}}} using spec:[{"operation": "shift","spec": {"FatturaElettronicaBody": {"DatiGenerali": {"DatiDDT": {"*": {"NumeroDDT": "FatturaElettronicaBody.DatiGenerali.DatiDDT[&1].NumeroDDT"}}}}}}] but getting result as null. expected output:{"FatturaElettronicaBody": {"DatiGenerali": {"DatiGeneraliDocumento": {"TipoDocumento": "TD01","Numero": "126561","Data": "05.05.2023","Divisa": "EUR","ImportoTotaleDocumento": "9382.29"},"DatiDDT": {"NumeroDDT":{"086708},"NumeroDDT":{086704},"NumeroDDT":{086705},"NumeroDDT":{086706},"NumeroDDT":{086707"}}}}}    Regards,Pradeep

I can't connect to HDFS kerberized and SSL enabled though I generated a keytab and certificate, Any help?

I've been testing the phoenixdb library in Python with CDP Operational DataStore cluster. Following the example here:https://community.cloudera.com/t5/Community-Articles/Using-phoenixdb-to-connect-Cloudera-Machine-Learning-to/ta-p/310130I can make a connection and return a single row but the connection is very slow to establish and execute. I'm running phoenixdb from the python interpreter on the COD cluster gateway host. The timestamped python execution is below. Python 3.6.8 (default, Nov 16 2020, 16:55:22)[GCC 4.8.5 20150623 (Red Hat 4.8.5-44)] on linux>>> import phoenixdb>>> from datetime import datetime>>> opts = {}>>> opts["authentication"] = "BASIC">>> opts["serialization"] = "PROTOBUF">>> opts["avatica_user"] = "workload-username">>> opts["avatica_password"] = "workload-password">>> jbdc_connection_url = "https://cod--<clusterID>-gateway0.<envID>.<custID>.cloudera.site/cod--<cluste