Hi All,   We are getting the below error intermittently for some tables and we have hive 3.1 . I am passing a sample query like insert into table test as select * from some table in Nifi PutHiveSQL processor .    Hive target table is managed table and stored as avro format   Error : PutHive3QL[id=d1652e76-54f7-30cf-8dc8-b1934cee3c26] Failed to update Hive for StandardFlowFileRecord[uuid=7447e7f0-4d1e-40fa-9965-5c98f7d11341,claim=StandardContentClaim [resourceClaim=StandardResourceClaim[id=1618924236521-4, container=default, section=4], offset=154416, length=3631],offset=0,name=20210421_test_branch_info,size=3631] due to java.sql.SQLException: Error while compiling statement: FAILED: Execution Error, return code 1 from org.apache.hadoop.hive.ql.exec.StatsTask. MetaException(message:Cannot change stats state for a transactional table ra_test.test_branch_info without providing the transactional write state for verification (new write ID -1, valid write IDs null; current state {"BASIC_STATS":"true","COLUMN_STATS":{}; new state {"BASIC_STATS":"true","COLUMN_STATS":{}); it is possible that retrying the operation will succeed, so routing to retry: java.sql.SQLException: Error while compiling statement: FAILED: Execution Error,   Option tried : if we rerun the query for the same table through Nifi PutHiveSQL , then Nifi processor executes the query fine . sometimes it fails in the first run.   Hive version : 3.0 and Nifi 1.11.4 Please assist what option to be checked in Hive side on this error.     Thanks ... View more

Hi All,   I am getting error as org.apache.hadoop.hbase.client.RetriesExhaustedWithDetailsException : Failed 1 action IO Exception.   We have Nifi flow which ingests JSON data into Hbase table 'test' . For this we have used Hbase client Service (HBase_1_1_2_ClientServiceId) to connect Hbase via Nifi Hbase Processor.The client connection works fine for sometime when flowfiles are flowing through the PuthbaseJSON,FetchHbaseRow and PutHbaseCell nifi processors . So Hbase nifi processor running and after 4 to 5 hours , if there is new flow file pass through that nifi processor , we see above error and for that i am doing workaround like disbaling Hbase clientService and Enabling the Controller Service manually    Scenraio1 : If we establish the Hbase connection initially , we will be able to insert ,update and access the Hbase table.So this is working fine and we are keep on ingesting the JSON status into Hbase continuously. Scenraio2 : If we are trying to insert JSON message into Hbase table after two hours through same Nifi flow , then we are getting the above error . Work around to ingest the JSON , I disable the Hbase client Controller service then Enabling the service . So the flow start ingesting the latest JSON.   We are having Nifi version 1.11.4.      We have also found this issue https://issues.apache.org/jira/browse/NIFI-7831 Please help for the solution and complete workaround ( there should not be any manual intervention ) to address this issue.   Thanks  . ... View more

Hi All,   I have requirement where i want to fetch two rows of Hbase table into single flowfile/attribute JSON format . I could not find way to do that with FetchHbaseRow nifi processor . I could pass single row identifier in FetchHbaseRow nifi processor and get the value.   I have configuration details stored in Hbase for Nifi flow . I have two row configuration, one row having generic configs and another row has flowfile specific configuration.   Please suggest on this .     Thank you ... View more

Hi All,   I am trying to convert input JSON to Output JSON using Jolt Transformation . My flow," AttributestoJSON" gives some count of flow files and after that i am using "Mergecontent" to merge the flowflies into single flowfile JSON .The single flowfile json passed to jolt transformation.   Input flow file JSON   {"filename":"test1.csv","message":"HiveSQL query executed Sucessfull!","status":"Success"} {"filename":"test2.csv","message":"HiveSQL query executed Sucessfull!","status":"Success"} {"filename":"test3.csv","message":"HiveSQL query executed Sucessfull!","status":"Success"}   Output JSON should be : { "filename": "${main_file}", "ex_status": "${Overall_status}", "error": "${final_message}", "fl_status": [ { "file_name": "test1.csv.", "status": "Success", "message": "HiveSQL query executed Sucessfull!" }, { "file_name": "test2.csv", "status": "Success", "message": "HiveSQL query executed Sucessfull!" }, { "file_name": "test3.csv", "status": "failed", "message": "HiveSQL query executed Sucessfull!" } ] }   Here main_file,Overall_status and final_message values , i am getting as flowfile attributes and will include the vaues for that from the attributes . I am exploring JSON specification and did not get the desigred result yet . Please help with Jolt specification for above output JSON .   Thanks @Nifi@Jolt ... View more

Hi,   I am trying to extract the single attribute(dynamic attribute passing) value from a JSON file or TEXT file .   The config file format looks like below { "product.csv" :"INSERT INTO Product", "Wholesale.csv" :"INSERT INTO Wholesale" }   The above JSON can evolve based new sources files and can added many file names with respective hive queries . I have config file at shareserver and able to retrieve the into middle of the nifi flow as per the attached screen-print . I able to retrieve the attribute values using Evaluate JSON path with fixed number of attributes . But the attribute value can be changed and i can't configure the same attribute name in Evaluate JSON path processor . I understand that getting attribute value by dynamically passing attribute names not possible in nifi processor like JSON to Attributes . https://community.cloudera.com/t5/Support-Questions/Nifi-Dynamic-Attribute-Creation/m-p/198062#M160113     I know that below JSON path is wrong for product.csv and it should be like $.product.csv . Assume that before Fetchconfig , i will update the "filename1" attribute to product.csv or wholesale.csv or any filename in Updateattribute processor . Using that attribute , i need to get the value from the config file dynamically . I would need some work around to pass dynamic "filename" as attribute and extract the values for that attribute and store into next processor flowfile attribute or flowfile content . I am still new to Nifi expression language . Please assist . Currently i need to retrieve 1 attribute value passing dynamic attribute value like filename @MattWho  @Shu_ashu @Nifi  @NifiDev    ... View more

Can you please advise how long the generated Hive audit Tez logs present in the below REST API https://IPaddress:/api/v1/views/TEZ/versions/0.7.0.2.6.2.0-205/instances/TEZ_CLUSTER_INSTANCE/resources/atsproxy/ws/v1/timeline/TEZ_DAG_ID Suppose if i execute the hive query in cluster ,I want to understand the how long the corresponding DAG ID present in the Ambari server. I am looking to read each Tez Dag details for Hive query monitoring purpose. ... View more

i am trying to establish the connection to Apache Solr from Apache Spark Dataframe and retrieve records. but getting Null Pointer exception while loading Solr collection to Spark Dataframe. I am able to load Apache solr index in a Spark Dataframe separate program . But when i try to add/integrate the solr connection logic( val options = Map("collection" -> "entity_APPLICATION", "zkhost" -> "localhost/solr") val initsolrdf = sqlContext.read.format("solr").options(options).load into SparkStreaming flow , i am getting null pointer exception while connecting to Solr from dataframe. val options = Map("collection" -> "lily_entity_APPLICATION", "zkhost" -> "localhost/solr") val initsolrdf = sqlContext.read.format("solr").options(options).load initsolrdf.show() I am working on consuming messages from kafka and getting few fields with values and those fields i need to send to solr as filter to get require indexed values from Solr.So my code consumed kafka message and in the same code while initiating connection to Solr via Apache Spark Dataframe , i am getting error . When i try the connection settings in different program in the same project its working fine. Working connection : val initsolrdf = sqlContext.load("solr", Map("zkHost" -> "localhost/solr", "collection" -> "entity_APPLICATION")) initsolrdf.printSchema initsolrdf.registerTempTable("ProducrDeDupe") Below is the part of main code where i am consuming json and retrieving few fields with values to send as filter to solr index via Apache Spark dataframe. def parseJson(jsonString: String): String = { val json: Option[Any] = JSON.parseFull(jsonString); val jsonMap = json.get.asInstanceOf[Map[String, Any]]; val application_processing = jsonMap.get("application-processing").get.asInstanceOf[Map[String, String]]; val applicant = jsonMap.get("applicant").get.asInstanceOf[Map[String, String]]; val application = jsonMap.get("application").get.asInstanceOf[Map[String, String]]; val product = jsonMap.get("product").get.asInstanceOf[Map[String, String]]; val product = jsonMap.get("product").get.asInstanceOf[Map[String, String]]; val product_category = product.get("product-category").get.asInstanceOf[String]; val product_type = product.get("product-type").get.asInstanceOf[String]; val fullname = applicant.get("full-name").get.asInstanceOf[String]; val owner_id = application.get("owner-id").get.asInstanceOf[String]; val duplicate_flag = application_processing.get("duplicate-flag").get.asInstanceOf[Boolean]; I have not passed any columns with values to Solr as a filters but just trying to load solr collection into Spark dataframe below. /* Initiating connection to Apache Solr and checking any dupliacte product type already created for the Applicant */ val options = Map("collection" -> "lily_entity_APPLICATION", "zkhost" -> "localhost/solr") val initsolrdf = sqlContext.read.format("solr").options(options).load initsolrdf.show() if (initsolrdf.count() != 0) { flag = "Y"; } else { flag = "N";[enter image description here][1] } val flag_update = application_processing + ("duplicate_flag" -> flag); val updatedJson = jsonMap + ("application-processing" -> flag_update); // } I have atatched the error message screenshot . Could you please assist where i go wrong . ... View more

I am aware of Ambari has all hadoop system performance metrices and dashboards in Grafana . But I looking to construct my own kafka producers,kafka consumers and kafka brokers metrics in Kibana so i have to log the system performance logs into Elasticsearch.So how can we approach this idea to start it? I know something like Mbean which has the system stats values ..Using Mbeans can we create metrics of the kafka brokers,producers and consumers ... View more