Solved

Hi Team, I am new to Nifi servers.I am currently working on a Development project where i developed a job to fetch changes from Mysql databases and apply them in a warehouse. I have 70 similar but different databases and therefore 70 Nifi jobs. The final aggregated database is around 100 GB. and daily some 1GB changes from all databases. I am currently working on development environment and using a single server.Now when i move to production with 70 Nifi jobs running all the time, do I need to think of a Nifi cluster ? or it is fine with single server ? What factors decide cluster vs stabdalone ? Also How can I handle upgrades to Nifi in future, Is it better for me to use Docker ? or having a normal instance is better ? Please help me with your valuable knowledge..

Is it possible to select with the QueryRecord processor also child fields of an given JSON structrue? E.g. following JSON: { "name" : "Max", "id" : 1, "location" : { "city" : "Frankfurt" }} And I would do the following query: SELECT name, location.city FROM FLOWFILE But this throws an error: QueryRecord[id=a4744340-017a-1000-6944-73fe4d362d90] Unable to query StandardFlowFileRecord[uuid=706e8ae8-f76c-4ebb-9c65-c46cb7026c89,claim=StandardContentClaim [resourceClaim=StandardResourceClaim[id=1626251405764-2, container=default, section=2], offset=790026, length=67],offset=0,name=55b72cfd-5308-4e83-9304-166193e76478,size=67] due to java.sql.SQLException: Error while preparing statement [SELECT "name", "location.city" FROM FLOWFILE]: org.apache.nifi.processor.exception.ProcessException: java.sql.SQLException: Error while preparing statement [SELECT "name", "location.city" FROM FLOWFILE] Is there something that I'm doing wrong? Here is the configurati

I am trying to roll my own search_after pagination query by saving the last sort values to a db and then retrieving them on the next time I run the query.  Thing is if there are no sort values from the db this means this is a new query and I skip the search_after clause in my query. If I do retrieve a set of sort values from the db then I need to plug them into the query. I am having trouble with the EL statement. This is what I have:   { "query": { "bool": { "must": [ {"term": {"key1": "value1"}} ], "must_not": [ {"exists": {"key2": "value2"}} ], "filter": [ {"range": {"@timestamp": {"lte": "now-1w"}}} ] } } ${search_after:isEmpty():not():ifElse(', \"search_after\": "${search_after}"', '')}, "sort": [ {"@timestamp": "asc"}, {"value3": "asc"} ]}   The ${search_after} variable defined as  '[ts int milliseconds, "a key value string"]'  Including the sq brackets. The error I am receiving is:Unknown key for a VAL

Hi, I am running CDH 7.1.7p0.  I have a external ORC database on S3 bucket.  I did not configure this fs.s3a.server-side-encryption-algorithm in hadoop core-site.xml.  However, I will randomly get this error when running hive query.  Rerun the same query may success or fail with same error message.  ERROR : Vertex failed, vertexName=Map 14, vertexId=vertex_1684599244994_0012_7_02, diagnostics=[Vertex vertex_1684599244994_0012_7_02 [Map 14] killed/failed due to:ROOT_INPUT_INIT_FAILURE, Vertex Input: ss1 initializer failed, vertex=vertex_1684599244994_0012_7_02 [Map 14], java.lang.RuntimeException: ORC split generation failed with exception: java.lang.RuntimeException: java.io.IOException: Cannot find password option fs.s3a.bucket.sg-cdp.fs.s3a.server-side-encryption-algorithmThe database and query was part of the hive benchmark tool from github:https://github.com/hortonworks/hive-testbenchCan you help me to resolve this problem.   Thank you.&

Set "OAuth2 Access Token provider" in Invokehttp prosessor with grant type as client credentials as well as client id and client secret, thinking no need to Set Authorization Bearer token in this processor, the processor will dynamically set retrieved token with post request to get remote API response. But always to 401 error back. Not able to tell it is error from token provider or from remove API to send post request. However, if using Postman, in a post  request, I can do it successfully to get remote API response. So I am sure the Grant type, client id and clietn secret are correct. What I am missing to get it work? WIl be very appreciate if NiFi experts guide me to configure invokeHttp processor for oauth2 seeteing to resolve my problem?

Is there a way to convert use expression language to convert the current time to a specific timezone.  Here is what I have currently but that's GMT not GMT+10:00. ${now():format("yyyy-MM-dd'T'HH:mm:ss.SSSXXX", "GMT")}

I'm attempting to transition from using PutBigQueryBatch to PutBigQuery. However, the latter doesn't seem to evaluate the dataset and table attributes as expected.In the provided example, one processor has the values hardcoded, which works correctly, while the other processor relies on retrieving those values from the flowfile's attributes. However, it appears that the attributes are not being properly evaluated. The example flow (using Nifi 1.21.0):The flowFile used: The left PutBigQuery processor has hardcoded values, and it works:The right PutBigQuery processor fails because it seems it's using the EL definition but it's not evaluating it: Any help would be appreciated.

Hi, I am trying ro read from a json file and create flowfiles with the kews as attributes and one flowfile per value in the list. The json file looks something like this:{"key1": null,"key2": null,"list": ["foo", "bar"],"key3": 1} This is my execute script:  import jsonimport java.iofrom org.apache.commons.io import IOUtilsfrom java.nio.charset import StandardCharsetsfrom org.apache.nifi.processor.io import StreamCallbackfrom org.apache.nifi.processor.io import InputStreamCallbackflowFile = session.get() if (flowFile != None): stream_content = session.read(flowFile) text_content = IOUtils.toString(stream_content, StandardCharsets.UTF_8) json_data = json.loads(text_content) for i in json_data["list"]: flowFile2 = session.create() flowFile2 = session.putAttribute(flowFile2, "key3", str(json_data["key3"])) flowFile2 = session.putAttribute(flowFile2, "key1", json_data["key1"]) flowFile2 = session.putAttribute(flow

Hi,Please let me know how to connect the Google Drive in nifi and read the files.Which Processor to be used.Please help in this regards,Best Regards,Rana

Hi, we get the following Ranger error - maybe you can help me to fix it as soon as possible?! (We activated MIT Kerberos). Thanks in advance! How can I check that the password of the keystore file is correct? And where can I change it? Feb 17, 2020 4:29:56 PM org.apache.ranger.server.tomcat.EmbeddedServer startINFO: Provided Kerberos Credential : Principal = rangeradmin/pdeluh0004392.hub.deluh.example.com@RDDL.PROD.EXAMPLE.COM and Keytab = /etc/security/keytabs/rangeradmin.service.keytabFeb 17, 2020 4:29:56 PM org.apache.ranger.server.tomcat.EmbeddedServer$1 runINFO: Starting Server using kerberos credentialFeb 17, 2020 4:29:57 PM org.apache.coyote.AbstractProtocol initINFO: Initializing ProtocolHandler ["http-bio-6182"]Feb 17, 2020 4:29:57 PM org.apache.coyote.AbstractProtocol initSEVERE: Failed to initialize end point associated with ProtocolHandler ["http-bio-6182"]java.io.IOException: Keystore was tampered with, or password was incorrectat sun.security.provider.JavaKeyStore.engineL

Hi Team, I want to setup kerberos client in my Mac Laptop having MacOS Monterey (version 12.6.5). I have put the krb5.conf file at below paths./etc/krb5.conf/Library/Preferences/edu.mit.kerberosBut when I try to run kinit, i get gelow error.--kinit -kt /Users/banshidhar_sahoo/Desktop/POC_KEYTAB/test.headless.keytab test@EXAMPLE.COMkinit: krb5_get_init_creds: unable to reach any KDC in realm EXAMPLE.COM, tried 0 KDCs-- I have also set the ENV Variable as below:KRB5_CONFIG=/etc/krb5.conf But getting same error while doing kinit. Can you please suggest how to point to krb5.conf so that it can reach out to the correct kdc server. Regards,Banshi. 

I have two processes1. consumerkafkarecord --mergerecord  --puthdfs2. consumerkafkarecord --mergecontent--puthdfswhen I use process 1, I have files in ndfs on the output readable spark, database, python libraries without problems, but the files are not larger than 200mb, all different sizes, although 500mb is set, but it is not filledwhen I use process 2 with the same parameters in mb and the number of lines, the files I get are exactly 500mb, but these files do not open, not by spark, not by any database, not by python librariesquestion why?I also want a large file always 500mb and so that it can be read without problems as in process 1mergecontent settingsMerge Strategy Bin-Packing AlgorithmMerge Format Binary ConcatenationAttribute Strategy Keep Only Common AttributesCorrelation Attribute Name No value setMinimum Number of Entries 10000Maximum Number of Entries 1000000Minimum Group Size 100 MBMaximum Group Size 500 MBMax Bin Age No v

How do I restrict the access to an endpoint provided by HandleHttpRequest to specific users? I did setup successfully a HandleHttpRequest-HandleHttpResponse chain, with Client Authentication = "Need Authentication". I provided a key- and truststore and users with a client certificate trusted by CA cert in the truststore can access the endpoint. However, I do not want everybody in my company to access this endpoint, I want to restrict this to a small group of users (actually to one single server). For testing purposes I created an empty truststore and imported into this truststore only my client certificate. However, now I cannot access the endpoint anymore (ERR_BAD_SSL_CLIENT_AUTH_CERT). How do I configure the JKS truststore to only trust one specific client? Or how do I configure HandleHttpRequest processor? If this is not possible, are there any other possibilities?Could I somehow use a Nifi-REST-API-Token?Could I somehow authenticate with the help

hello cloudera community, we check in the graph "total_kudu_on_disk_size_across_kudu_replicas" and there are tables with 500GB with that, we need to know what is the recommended size for a kudu table?

Nifi provenance data is not visible and remains at 80% in searching for provenance events after upgrading from 1.13.0 to 1.15.3 Below are the error logs : 2022-04-05 23:51:22,905 ERROR [Index Provenance Events-2] o.a.n.p.index.lucene.EventIndexTask Failed to index Provenance Eventsorg.apache.lucene.store.AlreadyClosedException: this IndexWriter is closedat org.apache.lucene.index.IndexWriter.ensureOpen(IndexWriter.java:877)at org.apache.lucene.index.IndexWriter.ensureOpen(IndexWriter.java:891)at org.apache.lucene.index.IndexWriter.updateDocuments(IndexWriter.java:1468)at org.apache.lucene.index.IndexWriter.addDocuments(IndexWriter.java:1444)at org.apache.nifi.provenance.lucene.LuceneEventIndexWriter.index(LuceneEventIndexWriter.java:70)at org.apache.nifi.provenance.index.lucene.EventIndexTask.index(EventIndexTask.java:202)at org.apache.nifi.provenance.index.lucene.EventIndexTask.run(EventIndexTask.java:113)at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)at jav

Hi, does anyone knows whether there are any restrictions on using external shuffle services on CDP?For instance, are the following external shuffle services supported?:- uber rss- apache uniffle

Hi Team, I trying to configure Apache Nifi LDAP authentication, Here is my nifi.properties. -----nifi.security.autoreload.enabled=falsenifi.security.autoreload.interval=10 secsnifi.security.keystore=/var/tmp/mks/certs/keystore.jksnifi.security.keystoreType=PKCS12nifi.security.keystorePasswd=************nifi.security.keyPasswd=nifi.security.truststore=/var/tmp/mks/certs/keystore.jksnifi.security.truststoreType=JKSnifi.security.truststorePasswd=**************nifi.security.user.authorizer=managed-authorizernifi.security.allow.anonymous.authentication=falsenifi.security.user.login.identity.provider=ldap-providernifi.security.user.jws.key.rotation.period=PT1Hnifi.security.ocsp.responder.url=nifi.security.ocsp.responder.certificate=------ Here is my "login-identity-providers.xml" -------><provider><identifier>ldap-provider</identifier><class>org.apache.nifi.ldap.LdapProvider</class><property name="Authentication Strategy">LDAPS&

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.

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": { "*": {

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 ? 

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

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.