Q&A

I have been using multiple record oriented processors ConvertRecord, UpdateRecord, etc. in various parts of my flow. For example, my UpdateRecord processor takes about 16 seconds to read in a 30MB flowfile, add some fields to each record and convert that data to parquet.  I want to improve performance such that this takes a lot less time.  My infrastructure that I am working on currently is a 2 node e2-standard-4 cluster in GCP with a Centos7 operating system. These two instances have 4 vCPUs and 16 GB RAM and for each repository (content, provenance, flowfile) I have separate SSD persistent drives.  A lot of the configs in NiFi are the defaults but what recommendations would anyone recommend either from an infrastructure or NiFi standpoint to improve performance on these processors.   

Can I know how to fetch AWS secret manager credentials with NiFi? Which processor I have to use? Even though there is a controller called "AwsSecretsManagerParameterProvider" where do I need to use this? My requirement is to connect with the AWS hosted Database

 I'm new to exploring Cloudera services, and I have a question: to use the data warehouse service, what components do I need to install and what are the minimum configurations?I plan to install it on VMware vSphere. I've learned that it requires Cloudera CDP Cloud Base and CDP Cloud Data Service (to be deployed on OpenShift with a minimum of 5 nodes). In my case, how do I determine the number of CDP Cloud Base nodes required? Do I need to install OpenShift for CDP Cloud Base?

Hello,I'm trying to edit Yarn Queue Manager queues and it seems i'm not able to edit them, there is not the option to edit them and there are no information in the doc about it...it's supposed to have this options panelBut this is not my case.the only information found it is to set the queue manager service in yarn, that's all...Any other ideas? 

I have a problem when sending JSON data to Elastic from Nifi - Nifi puts [ square brackets ] around a JSON array, and so Elastic doesn't recognise the data. The current fix is to use a SplitJSON processor, which removes the square brackets, but this will become unworkable when the number of flow files gets a bit big in number. Is there a way of creating an array of JSON that is recognised by Elastic that does not contain square brackets?I am also using an older version of Nifi - Ver 1.12.1

Hi,Would need some information on how to connect to Kubernetes pods of DW, DE DF ML on AWS. Do we need to make some configurations etc, what are the ways that we can connect? Any docs /guide would be helpful. Thanks Wert

I am setting up a stand-alone CDP cluster. However, getting errors when doing the agents installation steps. I have used this command but it still error:  

Hello,   I've been stuck on this issue for weeks and am currently on a wit's end. For the past few weeks, I've been trying to get my Apache Nifi to be able to post messages to a MsTeams channel. I've set up a MsTeams webhook and tried to use the 'InvokeHTTP' Processor to initiate a POST request to the MsTeams webhook. I've also tried using 'ExecuteStreamCommand' processor to invoke a curl Command to send a message to my Webhook URL, but its not working. This is what's inside my curl_command.sh,   curl -H "Content-Type: application/json" \ -d "{\"text\": \"Hello World from curl\"}" \ <Webhook URL>   Any help would be greatly appreciated  

Embedded zookeeper is failing to start while setting up a three node cluster for NIFI.The error message indicates that there is an issue with starting the embedded Zookeeper server while starting NiFi on your three-node cluster. The specific error causing the problem is "Address already in use," which suggests that the port required by Zookeeper is already occupied by another process. How can I solve this issue?  

Hello, I was trying to install feast python library for which protobuf upgraded version is required. But protobuf 3.19.4 is part of available packages in ML runtime addons, so installing protobuf upgraded version is not shown in pip list command.  Any suggestion on upgrading libraries which are part of ML runtime addons ?

We have a cluster running OS opensuse 12sp5. I wonder if CDH 5.14.2 supports this OS version. I‘v seen the CDH release notes, but still want to make sure.

I've been stuck for so long, I don't know what to do. Help❤️ I have installed Hive using ambari,When I try to insert a piece of data into the table, I get a Tez error,Here's what I've done:hive services and Tez services are normal. The user name I use is root, and encryption-free login has been configured on all nodes.root has its own /user/root directory and /tmp/hive/root on hdfsroot is a hive administrator and has all permissions on related databases and tablesI use yarn application-list to verify that no tasks are runningI use yarn node-list to confirm that all nodes are RUNNINGI use hdfs dfsadmin-report to confirm that all nodes are Normal and in normal stateI use hadoop fsck confirmed /warehouse/tablespace/managed/hive/root.db/test is healthyMy question is detailed below:I have a list like this0: jdbc:hive2://hdp0:2181,hdp1:2181,hdp2:2181> desc test;INFO : Compiling command(queryId=hive_20230614090425_2f598338-0603-4ba5-b4a2-8b77796b6d1b): desc testINFO : Semantic Analy

Hi to all,hope someone can point me on the right direction.I have something like that:[ { "reference" : "PB", "date" : "20230613", "barcode" : "C337287V28490010", "shipment_number" : ""}, { "reference" : "PB", "date" : "20230613", "barcode" : "C337287V28490011", "shipment_number" : ""}, { "reference" : "PB", "date" : "20230613", "barcode" : "C337287V28490012", "shipment_number" : ""} ]I need to update only a specific record with a met condition.Speaking on a SQL language it should be something like that:UPDATE FLOWFILE SET shipment_number = '${Shipment_Number_Attribute} ' WHERE barcode = '${Barcode_Attribute}'Unfortunately this statement in QueryRecord doesn't work, maybe UPDATE not supported?Moreover is there any way to achieve the same with UpdateRecord? Many thanks for any help.Appreciate. 

I have a csv file which a new record comes and appends every 15 minutes. I need to extract only the latest record from the csv file and save as a JSON

Dear all, My NIFI version is 1.16.3. I am trying to fetch the status of some processors (NIFI-API) within NIFI it self using InvokeHTTP processor.My NIFI-API is protected and need authentication to be used. I have a technical user with username and password that could get a token from the token API endpoint:https://nifi.company.com/nifi-api/access/token.So I defined a controller service OAuth2AccessTokenProvider to generate access tokens. From next defined API endpoint, I want to get the proprieties of a NIFI processor by its IDhttps://nifi.company.com/nifi-api/processors/ae91429e-es3a-1eq9-fe53-b13680cff1e5I used for that the InvokeHTTP processor with a GET HTTP method and by linking the OAuth2 Access Token provider field to the previously defined OAuth2AccessTokenProvider controller service.But I am getting such error InvokeHTTP Failed to properly initialize Processor. If still scheduled to run, NiFi will attempt to initialize and run the

Hello Experts,My current platform is build based on below components.CDP 7.1.7Hive 3Spark 2.4.7DeltaLibraryHadoop 3.1 All our data tables are hive external tables in parquet format.At present we have pyspark streaming solution in place to read json data from a golden gate feed and process through pyspark and deltalibrary solution in cyclic fashion.For each cycle, when streaming data is processed on delta path, we merge the data from delta path to hive external path but during this update on hive path, if there is someone who performs a query on this table gets an error that HDFS I/O file not found error.To solve this, we wanted to perform automic update on hive external table such that read queries are not failed when updates happening. We have tried below ways but didn't work.1) spark.sql2) hive.sql3) df.write (using jdbc connection)4) Regular df.write with overwrite mode and overwrite = True None of these worked, however it looks like "insert into overwrite .. " query works

My data set is as below."IM43163","SOUTH,OFC","10-Jan-23""IM41763","John:comment added","12-Jan-23"CREATE EXTERNAL TABLE database.table1( `col_1` string, `col_2` string, `col_3` string)ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.OpenCSVSerde' WITH SERDEPROPERTIES ( "separatorChar"="," , "quoteChar"="\"") Since the record is splitted amont two rows therefore it is not loading properly and gives null values. The output I am getting is mentioned below.IM43163SOUTH,OFC10-Jan-23IM41763?????

Is there capability to perform a reverse DNS lookup in NiFi without executing an external script, process, or code? (i.e. without execute code permission)

I have a flow that fetches all files from a given directory—they could be gzip, zip or csv, with the gzip and zip files holding a single csv file. I then route on MIME type, decompress the gzip files, unpack the zip files, and then bring what are now ALL csv files back together. This is working. I then want to create a zip archive of all of these csv files, and MergeContent seemed like a good candidate (outside of using ExecuteStreamCommand to run zip from the OS). But no matter what I do, the results are inconsistent:The first time I ran it with default properties in MergeContent (and setting Merge Format to ZIP), it created a single zip file — I thought I was done!But on the second time, it created six zip files, so I realized I wasn't done, and started playing with propertiesChanged Maximum number of Bins to 1 - created five zip filesChanged Correlation Attribute Name to absolute.path, which is the same for all flow files - created four zip filesChanged Maximum Group Size to 50

Hi,We have huge and complex XML files. For example: 15-20 levels in XML tree structure, approximately 180 basic types and 200 complex types, 1 to many relations between nodes in XML tree structure.As the output we want to have tables in Hive or Impala and to use SQL to query this tables. Could you please advise how to that in the most effective way?Effective - that is reducing manual coding works. Best regards

Nifi UI is showing blank screen. Processors are not visible. This is happening for all users/Administrators

Hi, I'm trying to connect nifi instance to HashiCorp Vault. As per instructions, I added HashiCorpVaultParameterProvider and StandardHashiCorpVaultClientService to my instance. I configured both of them and managed to fetch secrets from Vault to nifi. Right after I fetched secrets I was asked to create respective nifi context to work with fetched secrets. I created one and used in a couple of my processor groups simply by selecting context in "Process Group Parameter Context" dropdown. Secrets were accessable inside processor groups and worked fine. Then I updated secrets on the Vault side, and according to instructions I should have refetched them again using "HashiCorpVaultParameterProvider" but fetch icon did nothing. So new secrets were not come. In order to achieve this I had to drop created context and all related references in process groups. Only after these actions I managed to fetch new secrets again. So, it looks wierd because every time when I update something on

I read data from Kafka with ConsumerKafka processor and send them to MergeContent processor because I want to merge each hundred thousand data in a file. Then send them to UpdateAttribute processor cause I want to change filename, after that send them to PutSFTP processor cause I want to send data to an external server. My problem is that the number of data coming from Kafka is 100 thousand per second. But after going through this route, the output sends about 28 thousand data per second to the server and the lag value in the Kafka consumer increases over time. My Vm has 14 cores CPU and 21g RAM. The number of topic partitions that read from Kafka is 30, so I set thread to 30 in Consumer Kafka processor. Do you have any suggestions and ideas how I can increase the speed to reduce and eliminate the lag? 

Hey there,  for the sake of simplicity, I try to break down my problem to the core. I have flowfiles and their content is plain text (without any schema) representing data read from files on the filesystem. Via site-to-site I send them on a remote NiFi instance and place them there. During that ETL pipeline I need to query some additional metadata, sometimes via SQL queries, sometimes just some logic derived from the data.  Most of this metadata is simply added as attribute as desired. The ExecuteSQL processor however, overwrites the flowfile content, for example. This is why I need to duplicate/fork the flowfiles at some point. One flow keeps the original data content, the other queries the additional data which gets written to the flowfile content.  How can I merge these corresponding 2 flowfiles each back together? I know the ForkEnrichment and JoinEnrichment pattern, but they only merge content, additionally the content needs to be in some kind of sche

Dear Experts,We want to learn how to perform batch processing on semi-structured JSON flow files in NiFi. We know that we can use a Jolt Transform Record, an Update Record processor, and a Query Record processor, for which we need to set reader and writer controller services. However, we are unsure how to set up reader and writer controllers for semi-structured data schemas.In the text below, we have a multi-line JSON flow file in which the "code" field is static and the "other" field is an unstructured nested JSON. We want to query the "code" field in batch mode and include the "other" field information in the resulting flow files for processing in subsequent processors. How should we set the reader and writer controllers to achieve this?Regards,{"code": "6", "other": {"feild1": "data1"}}{"code": "7", "other": {"feild2": "data2"}}{"code": "8", "other": {"feild3": "data3"}}