About srinivaspadala_

srinivaspadala_ · ‎10-23-2017

Hi Team , to pull the data from servers , we are running ssh commands using ExecuteStreamCommand processor . looks like some ssh commands/flowfiles are blocked in this processor due to no response from server(it is acceptable and common in server) . if it is the case ,this processor has to terminate the flowfile , but it is not and after sometime , ExecuteStreamCommand processor is hanging up . Queue is showing some no of flow files ,but could not able to clear the queue in Nifi UI . looks ExecuteStreamCommand processor is trying or locked in it . what will be the solution to overcome this situation even if it is unable to process or run the ssh commads . Error we got in Nifi-app.log : 2017-03-27 04:03:56,125 ERROR [Timer-Driven Process Thread-82] o.a.n.p.standard.ExecuteStreamCommand ExecuteStreamCommand[id=5bda45d9-6601-14f0-3a44-294b3f5d994e] Transferring flow file StandardFlowFileRecord[uuid=633b08a2-f84d-4dc7-9a8c-175c5bc13b2e,claim=StandardContentClaim [resourceClaim=StandardResourceClaim[id=1490600595096-9, container=default, section=9], offset=0, length=0],offset=0,name=jmbfis01.northamerica.delphiauto.net,size=0] to output stream. Executable command ssh ended in an error: 2017-03-27 04:03:56,125 WARN [Timer-Driven Process Thread-84] o.a.n.p.standard.ExecuteStreamCommand ExecuteStreamCommand[id=5bda45d9-6601-14f0-3a44-294b3f5d994e] Processor Administratively Yielded for 1 sec due to processing failure 2017-03-27 04:03:56,125 WARN [Timer-Driven Process Thread-84] o.a.n.c.t.ContinuallyRunProcessorTask Administratively Yielding ExecuteStreamCommand[id=5bda45d9-6601-14f0-3a44-294b3f5d994e] due to uncaught Exception: java.lang.IllegalStateException: Partition is closed 2017-03-27 04:03:56,126 WARN [Timer-Driven Process Thread-84] o.a.n.c.t.ContinuallyRunProcessorTask java.lang.IllegalStateException: Partition is closed at org.wali.MinimalLockingWriteAheadLog$Partition.update(MinimalLockingWriteAheadLog.java:945) ~[nifi-write-ahead-log-1.1.0.2.1.2.0-10.jar:1.1.0.2.1.2.0-10] at org.wali.MinimalLockingWriteAheadLog.update(MinimalLockingWriteAheadLog.java:238) ~[nifi-write-ahead-log-1.1.0.2.1.2.0-10.jar:1.1.0.2.1.2.0-10] at org.apache.nifi.controller.repository.WriteAheadFlowFileRepository.updateRepository(WriteAheadFlowFileRepository.java:210) ~[nifi-framework-core-1.1.0.2.1.2.0-10.jar:1.1.0.2.1.2.0-10] at org.apache.nifi.controller.repository.WriteAheadFlowFileRepository.updateRepository(WriteAheadFlowFileRepository.java:178) ~[nifi-framework-core-1.1.0.2.1.2.0-10.jar:1.1.0.2.1.2.0-10] at org.apache.nifi.controller.repository.StandardProcessSession.commit(StandardProcessSession.java:363) ~[nifi-framework-core-1.1.0.2.1.2.0-10.jar:1.1.0.2.1.2.0-10] at org.apache.nifi.controller.repository.StandardProcessSession.commit(StandardProcessSession.java:305) ~[nifi-framework-core-1.1.0.2.1.2.0-10.jar:1.1.0.2.1.2.0-10] at org.apache.nifi.processor.AbstractProcessor.onTrigger(AbstractProcessor.java:28) ~[nifi-api-1.1.0.2.1.2.0-10.jar:1.1.0.2.1.2.0-10] at org.apache.nifi.controller.StandardProcessorNode.onTrigger(StandardProcessorNode.java:1099) ~[nifi-framework-core-1.1.0.2.1.2.0-10.jar:1.1.0.2.1.2.0-10]

srinivaspadala_ · ‎07-26-2017

I am newbie for pyspark , i could not able to get pyspark exception handling in transformations . for example i am calling a function on each line of map transformation , i would like to handle few exceptions in that functions and log them . exampleof my code : .map(lambda eachone : ultility.function1(eachone , somesrgs)) , here in function1 i would like to handle exceptions . please provide an example for better understanding . thank you .

srinivaspadala_ · ‎04-04-2017

we observed there is folder with hbase table name is created under /tmp . there are _temporary folders , part-m files are there in those . what might be the create/storing data here ? Is there any data loss or any table corruption happens if we try to delete those temporary table directory ?

srinivaspadala_ · ‎03-09-2017

2017-03-08 17:39:56,598 INFO [B.fifo.QRpcServer.handler=40,queue=0,port=16020] shortcircuit.ShortCircuitCache: ShortCircuitCache(0x5d8b409d): could not load 1122567351_BP-1119343884-10.192.24.155-1480056022466 due to InvalidToken exception. org.apache.hadoop.security.token.SecretManager$InvalidToken: access control error while attempting to set up short-circuit access to /apps/hbase/data/data/default/HBASE_TABLE_NAME/28610685b5622e4352e32afa842b45b0/INFO/4962a35101d5419989bdcc417d8d85f3 at org.apache.hadoop.hdfs.BlockReaderFactory.requestFileDescriptors(BlockReaderFactory.java:589) at org.apache.hadoop.hdfs.BlockReaderFactory.createShortCircuitReplicaInfo(BlockReaderFactory.java:488) at org.apache.hadoop.hdfs.shortcircuit.ShortCircuitCache.create(ShortCircuitCache.java:784) at org.apache.hadoop.hdfs.shortcircuit.ShortCircuitCache.fetchOrCreate(ShortCircuitCache.java:718) at org.apache.hadoop.hdfs.BlockReaderFactory.getBlockReaderLocal(BlockReaderFactory.java:422) at org.apache.hadoop.hdfs.BlockReaderFactory.build(BlockReaderFactory.java:333) at org.apache.hadoop.hdfs.DFSInputStream.blockSeekTo(DFSInputStream.java:662) at org.apache.hadoop.hdfs.DFSInputStream.readWithStrategy(DFSInputStream.java:898) at org.apache.hadoop.hdfs.DFSInputStream.read(DFSInputStream.java:955) at java.io.DataInputStream.read(DataInputStream.java:149) at org.apache.hadoop.hbase.io.hfile.HFileBlock.readWithExtra(HFileBlock.java:679) at org.apache.hadoop.hbase.io.hfile.HFileBlock$AbstractFSReader.readAtOffset(HFileBlock.java:1412) at org.apache.hadoop.hbase.io.hfile.HFileBlock$FSReaderImpl.readBlockDataInternal(HFileBlock.java:1625) at org.apache.hadoop.hbase.io.hfile.HFileBlock$FSReaderImpl.readBlockData(HFileBlock.java:1504) at org.apache.hadoop.hbase.io.hfile.HFileReaderV2.readBlock(HFileReaderV2.java:439) at org.apache.hadoop.hbase.io.hfile.HFileBlockIndex$BlockIndexReader.loadDataBlockWithScanInfo(HFileBlockIndex.java:269) at org.apache.hadoop.hbase.io.hfile.HFileReaderV2$AbstractScannerV2.seekTo(HFileReaderV2.java:636) at org.apache.hadoop.hbase.io.hfile.HFileReaderV2$AbstractScannerV2.seekTo(HFileReaderV2.java:586)

srinivaspadala_ · ‎01-26-2017

Thanks for your guidelines @Matt I am sure each flow file is having grade and branch attributes . as you said , We are getting NULL(no value set) values not empty for branch and grade in my case ,We have 3 conditions in which data will be : 1. sometimes we are getting NULL value for grade , not for branch . 2.sometimes , NULL for branch ,some value for grade . 3. sometimes both fields are getting NULL values . ex : ename, eno , salary , branch ,grade srini,1,10000,branch-A, sai,2,2000,,grade-A sai,3,, As i understood , conditions are wokr with 'AND' caluse , will above solution work for my case ? Thanks in Advance .

srinivaspadala_ · ‎01-26-2017

IS there any way to assign a default value if an attribute of a flowfile is NULL or empty in Nifi flow . ex: We are getting realtime streaming data (ename, empNO , salary , grade,branch) , now and then We are getting null values for the grade and branch fields . We would like to assign 'nograde','branch-none' values to grade,branch attributes in Nifi processor . Please let us know the way . Thanks in advance .

srinivaspadala_ · ‎12-15-2016

Hi , Is it possible to create an email alert when there is no flow files ,means no data is going through a connection or a processor ? Please provide your suggestions . thanks in advance .

srinivaspadala_ · ‎12-14-2016

Is there any way to archive HBase data like "Hadoop Archive" ?

srinivaspadala_ · ‎11-05-2016

The below input and script is working in http://jolt-demo.appspot.com/#andrewkcarter2 , But it is not working in Nifi JolttransformationJson . This example is concatenation of firstname and lastname . I understood it is becuase of not having "Operation : modify-overwrite-beta" in existing Nifi - JoltTransformation plugin . Is it possible to get this function by custom ? If yes, Please provide the process to implement . or let us know the process to ultilize it . Thanks . @Yolanda M. Davis - Please through some light on this .Thanks . Input : { "data": [{ "IDF": "123", "FirstName": "George", "LastName": "Kosher", "PaymentInfo": [{ "Type": "ABC", "Text": "Soft", "Amount": 3 }, { "Type": "ABC", "Text": "Text", "Amount": 5 }], "PaymentCard": [{ "CardNumber": "12345", "CardType": "Credit" }, { "CardNumber": "56789", "CardType": "Credit" }] }, { "IDF": "456", "FirstName": "Mill", "LastName": "King", "PaymentInfo": [{ "Type": "ABC", "InstructionText": "Hard", "Amount": 6 }, { "Type": "ABC", "InstructionText": "Text", "Amount": 8 }], "PaymentCard": [{ "CardNumber": "12345", "CardType": "Credit" }, { "CardNumber": "56789", "CardType": "Credit" }] }] } Script : [ { "operation": "shift", "spec": { "data": { "*": { // data arrayf "*": "data[&1].&", // pass thru stuff "PaymentInfo": { "*": { "Amount": "data[&3].Amount[]", "Text": "data[&3].PaymentText[]", "InstructionText": "data[&3].PaymentText[]" } }, "PaymentCard": { "0": { "CardType": "data[&3].CardType" } } } } } }, { "operation": "modify-overwrite-beta", "spec": { "data": { "*": { // data array "Name": "=concat(@(1,FirstName),' ',@(1,LastName))", "Amount": "=sum" // should work on Jolt 0.0.24 } } } }, { "operation": "remove", "spec": { "data": { "*": { // data array "FirstName": "", "LastName": "" } } } } ] Output : { "data" : [ { "IDF" : "123", "Amount" : [ 3, 5 ], "PaymentText" : [ "Soft", "Text" ], "CardType" : "Credit", "Name" : "George Kosher" }, { "IDF" : "456", "Amount" : [ 6, 8 ], "PaymentText" : [ "Hard", "Text" ], "CardType" : "Credit", "Name" : "Mill King" } ] }

srinivaspadala_ · ‎11-04-2016

first of all thank you @Yolanda M. Davis for quick response . Correct me if i am wrong , the above solution may work only eliminate duplicate json records based on one field, but we have a senarios like eliminating duplicates based on multiple fields . in the below example domain,location,time,function,unit Please provide the scripts to process in jolt . Thanks . or I can say simply eliminate duplicate json files from array of json Input : [{ "domain": "www.google.com", "location": "newyork", "time": "CDT UTC-0500", "function": "PACK", "unit": "PACK_ESR" }, { "domain": "www.yahoo.com", "location": "newyork", "time": "CDT UTC-0500", "function": "PACK", "unit": "PACK_ESR" }, { "domain": "www.google.com", "location": "newyork", "time": "CDT UTC-0500", "function": "AOI_S1", "unit": "AOI_L31" }, { "domain": "www.google.com", "location": "newyork", "time": "CDT UTC-0500", "function": "ALIGN", "unit": "ALIGN2" }, { "domain": "www.yahoo.com", "location": "newyork", "time": "CDT UTC-0500", "function": "PACK", "unit": "PACK_ESR" }, { "domain": "www.google.com", "location": "texas", "time": "CDT UTC-0500", "function": "PACK", "unit": "PACK_ESR" }, { "domain": "www.hortonworks.com", "location": "newyork", "time": "CDT UTC-0500", "function": "ALIGN", "unit": "ALIGN2" } ] Desired output : [{ "domain": "www.google.com", "location": "newyork", "time": "CDT UTC-0500", "function": "PACK", "unit": "PACK_ESR" }, { "domain": "www.yahoo.com", "location": "newyork", "time": "CDT UTC-0500", "function": "PACK", "unit": "PACK_ESR", { "domain": "www.google.com", "location": "texas", "time": "CDT UTC-0500", "function": "PACK", "unit": "PACK_ESR" }, { "domain": "www.hortonworks.com", "location": "newyork", "time": "CDT UTC-0500", "function": "ALIGN", "unit": "ALIGN2" } ]

Online	Offline
Last Visited	‎10-23-2017 12:05 PM

Member Since	‎09-28-2016 10:04 PM
Last Visited	‎10-23-2017 12:05 PM
Posts	59
Kudos received	15

Cloudera Community

Nifi ExecuteStreamCommand Process error with Illeg...

pyspark exception error handling ?

/tmp/hbase-tablename loaded with data , what might...

Error : ShortCircuitCache: could not load - acce...

Re: default value if an attribute of a flowfile is...

default value if an attribute of a flowfile is nul...

Nifi alert or Email notification when no flow for ...

HBase Archive ?

Nifi JoltTransformationJson for Operation : modif...

Re: Nifi JoltTransformJSON to remove duplicate Jso...