Q&A

Hi, I would like to extract a text between two characters for Example @7D1172E5@ using the extract text processor. How do I have to set the setting to get the text between th @ ? Thanks for support 

Hi, I want to convert some of the field value like  "enrolled_at":"Sun Oct 17 15:58:58 IST 2021" -> DateTime(2021-10-17 15:58:58) and "is_deleted":"{0}" -> boolean(true or false) from below Json array to save converted data in db. How i can achieve it in NiFi. [{"id":234521,"tenant_id":1,"product_id":1,"version":"3","user_id":197172,"user_ref_id":"8TS-0516","org_id":910,"org_subscription_id":7294,"access_code":null,"order_ref_id":null,"access_code_id":null,"enrol_type":"Admin4","enrolled_at":"Sun Oct 17 15:58:58 IST 2021","subscription_start_at":"Wed Oct 13 15:58:58 IST 2021","subscription_end_at":"Sat Oct 23 15:58:58 IST 2021","product_launched_at":"Wed Oct 13 16:05:49 IST 2021","product_completed_at":"Wed Oct 13 16:15:44 IST 2021","learning_completed_at":"Wed Oct 13 16:15:44 IST 2021","subscription_status":"ACTIVE","learning_status":"COMPLETED","total_credits":null,"completion_certificate_id":3585,"ecard_id":35867,"certificate_number":"tocubg3sfbkzfgfqs

Hello there, I'm upgrading a Nifi cluster (managed by ambri) to v1.15.3, therefore I have to secure it and activate ssl. I followed the installation steps from the official documentations, generated certificates (using Ambari Certificate Authority), and configured the Node Identities in Ambari. Still I have the "Untrusted proxy" error when I try to reach Nifi web interface. Below is my configuration :Nifi hosts as declared in Ambari :nif1.mydomain.comnif2.mydomain.comnif3.mydomain.comI'm accessing them (ssh & https) using other FQDNs, which I used to generate the certificates :nif1-adm.mydomain.comnif2-adm.mydomain.comnif3-adm.mydomain.comauthorizers.xml <authorizers> <userGroupProvider> <identifier>file-user-group-provider</identifier> <class>org.apache.nifi.authorization.FileUserGroupProvider</class> <property name="Users File">./conf/users.xml</property> <prop

Hello, I am trying to load tables from Kudu to HDFS using spark2 and i have noticed that timestamp is off by 8 hours between Kudu and HDFS.    df=spark_session.read.format('org.apache.kudu.spark.kudu').option('kudu.master','dcaldd163:7051,dcaldd162:7051,dcaldd161:7051').option('kudu.table',"impala::DB.kudu_table_name").load()   df.write.format("parquet").mode('overwrite').saveAsTable("db_name.kudu_table_name")   I have tried to set the timezone locally for the session in Spark2 and still it does not solve the issue.    Can someone give a hint on how to solve this issue? 

Hello. Im currently running nifi and postgresql on docker compose. I have a problem setting up the DBCPConnectionPool controller. I keep getting the error message below. Below is my configuration for DBCPConnectionPool I have checked the db name and docker compose yml file but nothing seems wrong on that end. What could be causing the problem here? 

Hello. I am simply spliting a csv file into two based on their row values. and when I run it through the QueryRecord processor, I get the error below. What could be the possible fix for this? Sample csv is as follows  GradeNm   ,CattleBreedNm   ,CattleGenderNm   ,PriceC++            ,beef_cattle          ,Female                  ,10094C++            ,hanwoo               ,Female                  ,12001C++            ,beef_cattle          ,Male                       ,12704B++            ,milk_cow             ,Female            &nb

Hello. I am currently using Nifi and Postgres via docker compose and is currently working on a sample project to get used to the tools.I am given a sample CSV file with 4 columns (grade, breed, gender, price)my scenario is as follows. 1. Split the sample CSV file into 2. (rows that have null value and rows that dont. So in the case below, row 1, 4 will be in the same table, and 2, 3 in the same table)  Grade ,BreedNm ,Gender ,PriceC++ ,beef_cattle ,Female ,10094C++ ,milk_cow ,Female ,nullnull ,beef_cattle ,Male ,12704B++ ,milk_cow ,Female ,16942   2. Save each of them into Postgresql tables3. Also save the logs into a table, using ID and the sequence of the log table Below is what I tried 1. Split flowfile in two.2. On one side, only save rows without null values and save them into a table 3. On the other side, save rows with null values and also save them into a table &nbs

I am new to Nifi.My requirement is to trigger Nifi process group using external scheduling tool called Control M. I tried using shell script to start and.stop the process group using curl command. Process group will fetch data from text file and writes into a database but unable to determine when the process group gets completed because I could see status like Started, Running and Stopped but not Completed state. Struck with this issue and need your valuable inputs on this of how to determine all the records got inserted into database placed inside process group. 

Please, i have files i want to move from a Nifi to a folder in another machine. The  ip address of Nifi & the destination are 173.24.206.40 ,173.24.206.44 respectively, meaning they are on the same network. I specified 173.24.206.44/home/username/destination_folder as distination directory in PUTFile processor BUT the destination_folder is empty & there was no error. Thanks.

My back pressure threshold for number of objects is default 10,000 while the data size is 1G. However, I saw the number of objects in the queue went far beyond 10.1000 though the data size was beyond 1G. Does this mean back pressure is applied only if both number of objects and data size reach the thresholds respectively? Another question. Suppose a kafka consumer processor is connected to a split processor. If the split processor reaches back pressure thresholds, will the kafka consumer stop consuming messages? Thanks, Mark

I have enabled Auto TLS mode but it did not work, so I disabled it from UI but it didn't work.When I try to add new host to cluster it always checks for heartbeat for server manager on 7182 instead of 7180. Things which  tried 1. Checked in DB but could not find anything in configs table (Using embedded DB)2. Disabled auto TLS from UI3. Modified /etc/cloudera-scm-agent/config.ini file and commented use_tls= 1 also changed the server port to 7180 but every time I try adding new host it automatically changes these properties back to normal4. Recreated UUID file. Kindly suggest the solution to this problem. TIA 

In a kerberized environment, I am creating a session on Livy with a kerberos ticket of User X and giving proxyUser UserY.curl -v --negotiate -u : -X POST -H "Content-Type: application/json" -d '{"kind":"spark","proxyUser":"UserY"}' https://server-url:8998/sessions -k The response I get is that{msg":"User 'UserX' not allowed to impersonate ''UserY"} when I initiate a ticket with UserY andcurl -v --negotiate -u : -X POST -H "Content-Type: application/json" -d '{"kind":"spark","proxyUser":"UserY"}' https://server-url:8998/sessions -k Response I get:{"id":145,"name":null,"appId":null,"owner":"UserY","state":"starting","kind":"spark","appInfo":{"driverLogUrl":null,"sparkUiUrl":null},"log":["stdout: ","\nstderr: ","\nYARN Diagnostics:.....} 

we are using kafka 2.2.1 in cloudera manager with cdh6.3.2 when we try to run kafka producer: kafka-console-producer --broker-list ip-172-31-13-30.us-east-2.compute.internal:9092,ip-172-31-12-151.us-east-2.compute.internal:9092,ip-172-31-14-136.us-east-2.compute.internal:9092 --topic test_topic returns the errors: 22/07/08 12:36:16 WARN clients.NetworkClient: [Producer clientId=console-producer] Bootstrap broker ip-172-31-14-136.us-east-2.compute.internal:9092 (id: -3 rack: null) disconnected22/07/08 12:36:16 WARN clients.NetworkClient: [Producer clientId=console-producer] Bootstrap broker ip-172-31-12-151.us-east-2.compute.internal:9092 (id: -2 rack: null) disconnected22/07/08 12:36:16 WARN clients.NetworkClient: [Producer clientId=console-producer] Bootstrap broker ip-172-31-13-30.us-east-2.compute.internal:9092 (id: -1 rack: null) disconnected when we check the logs of kafka brokers there is: 2022-07-08 12:36:17,365 INFO org.apache.kafka.common.network.

I am getting String Date "Wed Oct 13 15:58:58 IST 2021" in changeCaptureMySql processor flowfile content out claim which is getting read from mysql-bin.log file as source DB data. Destination DB column has datatype DATETIME. I am using the JsonPathReader and AvroSchemaRegistry to read the each column data to save in db but it is giving data incorrect date-time value error. Q1. Is there any away I can convert String date time value to DATETIME to save it without error?Q2. How to define the DATETIME data type in AvroSchemaRegistry ?. I tried below data type in AvroSchemaRegistry but it won't work.Please find the attached screenshot for reference.{ "name": "TIMESTAMP" , "type": { "type":"long", "logicalType":"timestamp-millis"} }, { "name": "DATE" , "type": { "type":"int", "logicalType":"date"} } 

Hi , I am trying to automate data from 2 source DB to one destination DB where some the tables name of source DB is same but in destination DB I have given table name as soursceDBName_tableName. QA1-> How I can define common "table names" in changeCaptureMySql processor configuration when 2 source database has same tableName and columns and destination DB have different  tableNames ?QA2-> Also how i can identify these tables that which data source it comes to save in right table in destination DB Ex. subscription DB(sourceDb1) -> status(tableName)userOrg DB(sourceDb2) -> stats(tableName)reportDb (destinationDB)  -> subscription_status(TableName), userOrg(TableName). Please refer the screenshot for same.  

Hi,How to insert or update a record in mySQL table if sourse db table name X and destination db table name Y using UpdateAttribute processor. I am able to achieve this if source and destination has single table but I have multiple table where source db table's names A,B and destination table's names Y,Z . I tried below condition but getting a error.schema.name : ${${databaseName}:equalsIgnoreCase(nrpsubscriptiondb2306):and(${tableName}:equalsIgnoreCase(status)):ifElse(subscription_status,${tableName})}   Error log: due to Unexpected token ':' at line 1, column 76. Query: ${${databaseName}:equalsIgnoreCase(nrpsubscriptiondb2306):and(${tableName}:equalsIgnoreCase(status)):ifElse(subscription_status,${tableName})}    

Hi guys, i've got a queue containing ~1000 files and i want to receive a mail with all ~1000 filenames. I don't care about the content of all files, only the ${filename} should be put on a list and send as content. Example: - Queue Position 1 - filename: first.xml- Queue Position 2 - filename: second.xml- Queue Position 3 - filename: third.xml Resulting mail content (possibility 1):- first.xml- second.xml- third.xmlResulting mail content (possibility 2):first.xml, second.xml, third.xmlSomething just like a list of filenames in queue.Dies anybody got an idea how to solve without many api-magic?Greetings and best regards - Max

Dear all, I have a question on using a new set of data sources on the currently installed cloudera CDP platform in Ooredoo network. Currently the OPEX optimizer uses Telco core and network access along with Charging System Data sources. Now, we have an opportunity to add a new set of IT (non Telco) data sources that may contain offline batch data sources as well as streaming data (ex. For Fraud Management and Revenue Assurance depts), shall we : Install a new Cloudera CDP platform fully dedicated to the new data sources (IT data source).Or use the existing CDP platform ? If the second option is chosen, what are the components and tools that will be used for ingestion, store and process the new data sources, and what are the nodes that will be implied to implement them ? can we dedicate some worker nodes working only for IT data sources for the same HDFS ? how is the configuration of the components and tools then ? Best Regards, mamoune,

Hi,  I have to work on nifi data flow to migrate data from one database to another database from multiple schemas. I have to migrate only specific set of tables with a filtering applied to each of the tables. I came up with a sample flow where i used one ExecuteSql Processor for each table that has to be migrated. In the Sql query for the processor i applied the filter criteria  and then connected that processor to an individual PutDatabaseRecord processor for each destination table.  Just want to understand is that the right approach to have separate processors for each source and destination table? Also would appreciate some guidance on what are the best processors to be considered for data migration from one db to another?   This will be a continuous migration to be used in Production environment that will keep running for a couple of months.  Appreciate any help.  Thanks   

Hi,I am using a custom python script (image attached below) in ExecuteProcess processor and it generates flowfiles as result of stdout. The problem I am facing is that for all stdouts, it generates only one flowfile. My requirement is to generate seprate flowfile for each stdout.I tried putting a pattern in stdout and feeding the output flowfile to SplitContent as input. But I keep getting "A flowfile is currently penalized and the data cannot be processed at this time" and it doesn't resolve even after waiting for so long. SplitContent doesn't show any error either, it just doesn't process the splitting.Trying to split the flowfiles based on pattern - "#;#" [PFA the images below]I would really appreciate any help! PS.: Also tried to use ExecuteScript to run custom python script, but I guess since it works on jython, it didn't allow me to import some python packages. 

Is there a way to use SSO without enabling HTTPS on NIFI?

Hi, I'm getting this message when running process with more than 10000 flowfiles.unable to write flowfile content to content repository container default due to archive file size constraints I googled and read more about content repository and content claims, but still don't get it.My archive folder is like 100MB. Can you please help me understand how to configure NiFi to get rid of this message? Thanks a lot. My settings which could affect this is:nifi.flow.configuration.archive.max.time=30 daysnifi.flow.configuration.archive.max.storage=500 MBnifi.flow.configuration.archive.max.count= nifi.queue.backpressure.count=10000nifi.queue.backpressure.size=1 GB nifi.content.repository.archive.max.retention.period=12 hoursnifi.content.repository.archive.max.usage.percentage=50%nifi.content.repository.archive.enabled=true 

Hello all,I am installing Data Services with Embedded Container Service. There are 2 dedicated servers: 1 ECS server, 1 ECS agent. First installation run not good, always stuck at this task:Waiting for process rke-wait+ kubectl taint node -l ecs_role=master node-role.kubernetes.io/control-plane-taint "node-role.kubernetes.io/control-plane" not foundtaint "node-role.kubernetes.io/control-plane" not found+ true+ sleep 10+ /var/lib/rancher/rke2/bin/kubectl --kubeconfig /etc/rancher/rke2/rke2.yaml get nodes -o 'jsonpath={range .items[*]}{@.metadata.name}:{range @.status.conditions[*]}{@.type}={@.status};{end}{end}'+ grep Ready=True++ hostname -f+ grep instance-20220615-1150.subnet06081614.vcn06081614.oraclevcn.com+ echo ...+ [[ ! -z true ]]+ kubectl taint node -l ecs_role=master node-role.kubernetes.io/control-plane-taint "node-role.kubernetes.io/control-plane" not foundtaint "node-role.kubernetes.io/control-plane" not found+ true Anyone face this familiar issue before? how can i fix

I am trying to import an *.xml template made with NiFi version 1.13.2 into a 1.14 version client (in docker).When trying to upload the *.xml file with the Upload Template button under the Operate panel, I get the following error message: <html> <head> <meta http-equiv="Content-Type" content="text/html;charset=utf-8"/> <title>Error 403 Forbidden</title> </head> <body><h2>HTTP ERROR 403 Forbidden</h2> <table> <tr><th>URI:</th><td>/nifi-api/process-groups/8f8e29ed-0181-1000-d7d0-76467ebfce01/templates/upload</td></tr> <tr><th>STATUS:</th><td>403</td></tr> <tr><th>MESSAGE:</th><td>Forbidden</td></tr> <tr><th>SERVLET:</th><td>jerseySpring</td></tr> </table> <hr><a href="https://eclipse.org/jetty">Powered by Jetty:// 9.4.42.v20210604</a><hr/> </body>

Hi, We are having an issue reading tables in Impala with datatype Decimal (1,1). The error is: column 'DECIMAL_1_1' has an invalid type length. Expecting: 1 len in file 4   We have two Cloudera clusters (CDH 6.3.3). This error only occurs in the 2nd cluster. We write data into Parquet files through IBM Datastage. I run the same job in both clusters to examine the parquet files schema using parquet-tools schema , and I found this: In the 1st cluster, in the parquet file the schema of the column causing the issue is:optional int32 DECIMAL_1_1 (DECIMAL(1,1)); However, in the 2nd cluster:optional fixed_len_byte_array(4) DECIMAL_1_1 (DECIMAL(1,1));  We need to understand why the type is fixed_len_byte_array(4) in this parquet while it's int32 in the other. Apparently the fixed_len_byte_array(4) is causing the issue with Impala.