Solved

Hello,let’s say a Datalake is used by several departments within a company.What are recommended ways to measure Compute and Storage by department, for billing purposes?Many thanks.

We’re working on securing NiFi and I’m coming across an issue where I intermittently get an exception. The environment is a two-node NiFi cluster which is configured to authenticate users using LDAPS, and authorise them using the FileAccessPolicyProvider / FileUserGroupProvider which we plan to replace with LDAPS. The login looks to be working. I think this is the authorisation part which is failing (although it works sometimes - before cluster voting completes). We have configured a certificate trust store for the CA (which is an internal CA), and a key store for the SSL cert. Note: We're using a san cert for SSL. Exception is… 2018-01-19 08:41:23,999 WARN [Replicate Request Thread-9] o.a.n.c.c.h.r.ThreadPoolRequestReplicator Failed to replicate request GET /nifi-api/flow/current-user to 10.101.50.5:8443 due to com.sun.jersey.api.client.ClientHandlerException: javax.net.ssl.SSLHandshakeException: Received fatal alert: certificate_unknown 2018-01-19 08:41:23,999 WARN [Replicate Reques

Good Morning,I'm flummoxed and worried. I have a significant flow that handles parquet files, and I can no longer make it work in version 2.0. Specifically, I used a ConvertRecord processor with a ParquetReader and a JsonRecordSetWriter.In the new version, 2.0, There doesn't appear to be a ParquetReader, anymore. There also doesn't appear to be any documentation about the ParquetReader being deprecated.Please help. This flow is critical to me. I am also open to suggestions on how to accomplish the same conversion in a different manner.

Attempting to add a new queue using YARN Queue Manager UI.  After saving, the following error message appears: Validation failed for Add queue operation! Error message:<?xml version="1.0" encoding="UTF-8" standalone="yes"?><RemoteException>    <exception>AuthorizationException</exception>    <message>Unable to obtain user name, user not authenticated</message>    <javaClassName>org.apache.hadoop.security.authorize.AuthorizationException</javaClassName></RemoteException> Environment:  CDP DC 7.1.4Kerberized:  YesTLS:  NoLDAP/AD:  Yes Checked Ranger logs and Audit logs, Yarn RM and security logs.  Couldn't find any related errors in those log files.  The Ranger Audit logs show the user accessing the YARN queue but does not show a denied result.  There are LDAP/AD related errors in Usersync...would that cause authentication issues for YARN Queue Manage

There is a bug with the InvokeHTTP processor in version 2.0. I carried over my flows from version 1.23, which I upgraded to 1.24, and then I exported the flows as processing groups and imported them into version 2.0.  These flows worked fine, previously. In one, I have an InvokeHTTP processor that is sending the document to a Tika server for processing.  In version 2.0 there is a new property in this processor that was added. I did not put it there. The documentation makes no mention of it.The property was `Proxy Type` and it's value was `http`. I tried to set it as an empty string, but I continued to get a failure. The error I was getting was:18:17:26 UTC ERROR InvokeHTTP[id=dfae880c-b84a-3f2c-f1d9-941b8fc625eb] Request Processing failed: FlowFile[filename=random_20210909a.docx]: java.lang.IllegalArgumentException: Unexpected char 0x20 at 5 in header name: Proxy TypeEven after setting this property to an empty string, the error persisted. I remedied the situation by del

Hi all, I have built an Apache nifi cluster consisting of 5 nodes and a separate zookeeper server. After that I login to webui and use the ListenUDP processor. This processor opens for example 50000 port on all nodes and it can receive information on any node. For example I will send information to host1:50000, then it should distribute the information on all nodes (host2, host3, host4, host5). It does all this and everything works, but I have two questions.1) Does host1 participate in sending logs further down the processors or does it just receive and distribute?2) The Nifi cluster is called fault tolerant, but what if host1 becomes unavailable? Essentially it will not be able to receive information and distribute it to nodes. Is there any solution for this? So that when host1 goes down, the information can be transferred to host2?

at first, i want to reuse my model in HDFS, but it reported something wrong:it seems to be because the spark version is not set up correctly, I read the source code and found that spark calls sc.version when it stores the model. However, embarrassingly, the spark2.4.0 that comes with CDH6.3.2 that I use produces the following result: SPARK_VERSION is empty.   I need to use CDH spark to save and read the model, what should I do in this case, thanks very much!

Hey there,I'm trying to create a nested JSON using SQL in NiFi, but the output I'm getting has the nested part as a string. Here's the query I'm using:SELECTorder_id,JSON_ARRAYAGG(JSON_OBJECT('order_Item_Seq_Id', order_Item_Seq_Id,'product_Id', product_Id)) as order_itemFROM order_itemGROUP BY order_id;The output looks like this:[{"order_id": "10000","order_item": "[{\"product_Id\": \"10007\", \"order_Item_Seq_Id\": \"00101\"}]"}]But I want it to be a valid nested JSON like this:[{"order_id": "10000","order_item": [{"product_Id": "10007","order_Item_Seq_Id": "00101"}]}]I'm hoping for a solution using Apache NiFi or Jolt.Any help would be appreciated.Thank you!

Hi guys! I have json like that:{"id" : "20","user" : {"name" : "John","date" : ""}}I transform this date to Long using nifi langugage ${date:isEmpty():ifElse('',${date:toDate('yyyy-MM-dd HH:mm:ss'):toNumber()})} and after this i do in jolttrasformation something like[{"operation": "shift","spec": {"idTransakcji" : "id","date": "user.date",}},{"operation": "modify-overwrite-beta","spec": {"user": {"date": "=toLong(@(1,date))"}}}] and it works. But I have problem beacause when I have emptyString in this date i need to transfer it in null(not null in String) and i dont know how to do it using JoltTransformation or something else  so as not to lose what I already have and have null only when there is an empty string. 

Hi everyone,I'm currently encountering an issue related to passing request bodies in my workflow. Here's a breakdown of the situation:In my workflow, I have a total of 9 processors, all of which are InvokeHTTP Processors utilizing post requests. For each post request, I need to include a request body.The challenge lies in figuring out how to effectively pass the body using the InvokeHTTP processor. Presently, I am unable to accomplish this by utilizing the InvokeHTTP processor directly. We've attempted using GenerateFlowFile to pass the body, but it cannot be seamlessly integrated into the workflow. Specifically, I am unable to insert GenerateFlowFile between Processor 3 and Processor 4, for example.One potential solution we've considered is creating individual GenerateFlowFile instances for each processor outside the workflow. However, this approach introduces complications, especially when one processor is dependent on the output of a previous processor. For instance, if Processor 5

I was trying to find some documents about where would the location for HS2 & HMS logs be located for CDP-PC on AWS, but could get none the nearest I could get is https://docs.cloudera.com/data-warehouse/cloud/troubleshooting/topics/impala-logs-s3.html. but this is not what I want. Appreciate any help in this.

Hi,We are currently working on scenario to connect to secured Rabbitmq server for consuming messages. We have used StandardSSLContextService for that purpose and configured both truststore and keystore properties.  These certs are created specific to Rabbitmq connection.Everything is working fine as expected but the only issue we are facing is that when both keystore and truststore are expired and we need to update the certs currently the only way we see is to restart the server or controller service.But we would like to know if there is any other way the NIFI StandardSSLContextService can check if the cert is expired and if we provide a new keystore and truststore then it will reload the latest certs without the need for restarts.Any suggestions on this issue is greatly appreciated.

Hi ,iam retrying a flow file 3 times before moving moving it to failure relationship  , with each retry i want to have 10 seconds gap in between them , so 2nd retry will happen after 10 seconds .iam not able to understand how to use wait processor for this . What should be the configurations of that processor .iam using older version of nifi so i can not use retryFlowFile processor . 

Hi ,is there any idle connection timeout for ConsumeKafka and PublishKafka processors in NIFI.Do these processors go to idle state if no data in incoming ??if yes then what is the time in which they will go to idle state in case of no data is coming ??? and is there any way to set this time ourself ???

Hi Guys, I would like to know how to set up authentication for spark history server, so that unauthorized users cannot view the spark history server web ui, any help would be appreciated, thanks! 1. I do notice there are below statements in the official spark document:“Enabling authentication for the Web UIs is done using javax servlet filters. You will need a filter that implements the authentication method you want to deploy. Spark does not provide any built-in authentication filters.” 2. There is also a thread on stackoverflow regardig this : “You re-use Hadoop's jetty authentication filter for Kerberos/SPNEGO: spark.ui.filters=org.apache.hadoop.security.authentication.server.AuthenticationFilter and spark.org.apache.hadoop.security.authentication.server.AuthenticationFilter.params=type=kerberos,kerberos.principal=${spnego_principal_name},kerberos.keytab=${spnego_keytab_path}”. with kerberos authentication enabled in CDH6.3, I followed the instruct

Hello everyone,I'm seeking assistance from those experienced in NiFi for a specific challenge. I use NiFi for internal tools and simple integrations within my company, organizing different functionalities into separate ProcessGroups. These functionalities include converting data types (e.g., from JSON to CSV), generating dummy data, and sending data to a partner through an API.To make all ProcessGroups accessible through an endpoint, I've implemented a setup with a "HandleHttpRequest" processor listening on a specific port in the main canvas. Subsequently, a "RouteOnAttribute" processor is used to route requests to the appropriate ProcessGroup based on the URL path.However, this approach has its drawbacks. The canvas becomes cluttered, requiring encapsulation of flows within more general ProcessGroups. Additionally, every time I add a new flow, I have to stop the "RouteOnAttribute," update it, link the new relationship, and then restart it (I know I can use NiFi API to accelerate this

Hi guys! i have some problems with JoltTransformation in NIFI. I have plain json like that:{"id": "c47f79f5-77b1-45d1-91df-968c2fa3def8","system" : "OP","sourceSystem" : "DKG","name" : "Agnes","lastName" : "Smith","mail" : "a.smith@gmail.com","country" : "Australia","city" : "Sydney","street" : "Colorado","building" : "23","local" : "1',"paycheck" : "3456,98',"joiningDate" : "12-09-2023","isOfAge": "true",}And i need to have output like this:{"paycheck":3456.98,"joiningDate":1702048486451,"person":{"name":"Agnes","lastName":"Smith","mail":"a.smith@gmail.com","isOfAge":true,"adress":{"country":"Australia","city":"Sydney","street":"Colorado","building":"23","local":"1"}},"system_data":{"id":"c47f79f5-77b1-45d1-91df-968c2fa3def8","system":"OP","sourceSystem":"DKG"}}In addition to nesting, I need to convert date to long, paycheck to double and isOfAge to boolean.I tried use JoltTransformation and Jolt Specification like this:[{"operation":"shift","spec":{"paycheck":"paycheck","joiningDate"

Hi,we would like to activate HA for Cloudera CDP MySQL databases (hive, ranger, etc..).We have seen that there are several ways to achieve all this, mysql replication active-slave or active-active, mysql + DRDB, mysql Innodb Cluster, etc..Our goal is to achieve automatic failover of mysql, with functionality similar to those of a cluster with an active node and a standy node.From your experience, is there a recommended architecture for creating HA for a Mysql server compatible with Cloudera services? I read that some solutions could not be compatibile with Hive metastore for example.Thanks in advance.Ivo

Hi, Guys. I am new to use NiFi flow. I am kind of confused about Query database table processor. So now, I am try to get all data from Postgres database and convert them into json and then convert them into xml.But sometimes, the Query database table processor will have a lot of Flow file and then the queue just mess up. I try to set the art_id to be maximum-value column but it doesn't work well because art_id is string.(not sure)This is a screenshot from a sample table called title and it has about 500 rows. How to query them just one time and split them into 500 flowfile and each of flowfile has only one row from table title without any duplicates. Can someone help me?Thank you so much. Following screenshots are my progress so far.     

Hii Everyone, I've been struggling to achieve the desired result with a Jolt transformation for the past few days and could really use some help from the community. input.json: { "email": "abcmail@gmail.com", "dates": [ { "date": "2023-12-01", "type": "full" }, { "date": "2023-11-03", "type": "full" }, { "date": "2023-11-02", "type": "full" }, { "date": "2023-10-01", "type": "half" } ]} output.json [ { "emailId": "abcmail@gmail.com", "date": "2023-12-01", "type": "full" }, { "emailId": "abcmail@gmail.com", "date": "2023-11-03", "type": "full" }, { "emailId": "abcmail@gmail.com", "date": "2023-11-02", "type": "full" }, { "emailId": "abcmail@gmail.com", "date": "2023-10-01", "type": "half" }] spec file for the transformation: [ { "operation": "shift", "spec": { "*": { "email": "[].emailId", "dates": { "*": {

Hello guys,I'm struggling with JsonRecordSetWriter 1.17.0, we have valid nested JSON data, but for some reason whenever this data passes through JsonRecordSetWriter, on nested arrays it adds the following :   "orderItemsCustomizes" : [ "MapRecord[{quantity=1, itemCurrency=AED, totalPrice=0, orderItemId=2767, itemCustomizeGroup=MapRecord[{itemModifierGroupAttributes=[Ljava.lang.Object;@4658e00e, customizeGroupName=Burger Selection}], itemCustomize=MapRecord[{itemCustomizeGroupId=2308, itemId=18622, itemName=Chicken Burger, isAvailabel=true, updated_at=2023-01-27 15:59:31, itemCurrency=AED, itemCost=0.00, created_at=2023-01-27 15:59:31, itemModifierAttributes=[Ljava.lang.Object;@ee87ab1, id=7539, itemImage=https://d2cvcbugmdflrn.cloudfront.net/e777bb72-a445-4162-9844-f9bda5e7a3d0.jpg}], itemId=18622, itemCustomizeGroupId=2308, createdAt=2023-01-27 16:48:02, itemCustomizeId=7539, price=0.00, ID=1153, updatedAt=2023-01-27 16:48:02}]", "MapRecord[{quantity=1, itemCurrency=AED

Hi ,In Apache NIFI  I am  consuming data from one Kafka topic and publishing that data to customers Kafka topic . For this i am using ConsumeKafka and PublishKafka processor respectively.If outgoing dataflow is idle for more than 5 minutes, the Kafka connection with customer should be terminated and  reconnected when there is  any new  messages consumed .how to do this with publish kafka processor ?? how to terminate and start the connection??

I am using ExecuteSQL to pull a colb column from DB. ExecuteSQL provides the result in the avro format. Then I am converting that to Json using AvroToJsonProcessor but it is adding escape characters. I want the value that is being pulled from DB as a Json but I am getting it as a string with escape characters. [    {        "SEARCH_RESULT": "{ \"requestId\": \"203680\", \"responseData\": [ { \"searchKey\": \"cardNumber\", \"data\": [ { \"firstName\": \"Martin\", \"lastName\": \"Garry\" }, { \"firstName\": \"Martin\", \"lastName\": \"Garry\"}, { \"firstName\": \"Martin\", \"lastName\": \"Garry\",  } ] } ] }"    }]I want that value of Search result to be a Json instead of a String so that I can use JOLT to transform it as needed. Can you someone pls suggest how to do it.

Hello,I'm running a sqoop job through Hue. This imports data from Oracle, but the value of --query is being divided by whitespace and produces sqoop flag error because it does not recognize these parametersimport --connect jdbc:oracle:thin:@server/service --query 'select * from mytable'The logs show somethng like this:Sqoop command arguments : import --connect jdbc:oracle:thin:@server/service --query 'SELECT * from mytable'-------------------------------------------------------------------------------------------------------17:48:29.329 [main] ERROR org.apache.sqoop.tool.BaseSqoopTool - Error parsing arguments for import:17:48:29.329 [main] ERROR org.apache.sqoop.tool.BaseSqoopTool - Unrecognized argument: select17:48:29.329 [main] ERROR org.apache.sqoop.tool.BaseSqoopTool - Unrecognized argument: *17:48:29.329 [main] ERROR org.apache.sqoop.tool.BaseSqoopTool - Unrecognized argument: from&nb

hi everyone ! hoping everybody has an awsome day.I am trying to create a process Group with a flow definition json i downloaded.when i upload a file.json manually and checked the nifi-request.log and the only i see is this pettiotion " http://localhost:8081/process-groups/{id}/process-groups/upload" but i don't see the body in sent request, i want to use the api rest to upload this file but documentation is not clear in this aspect pettition #1:curl --location "http://localhost:8082/nifi-api/process-groups/{id}/process-groups/upload" \--form "file=@/path/to/my/flowDefinition.json"Response: Process Group name is required pettition #2:curl --location "http://localhost:8082/nifi-api/process-groups/{id}/process-groups/upload" \--header 'Content-Type: multipart/form-data' \-F 'id=root' \-F 'name=QA_FLOW' \-F 'position[x]=577.0' \-F 'position[y]=11.0' \-F 'clientId=27bf6df2-018c-1000-362e-61166913eee4' \-F 'disconnectNode=true'-F "file=@/path/to/my/flowDefinition.jso