About te04_0172

te04_0172 · ‎12-03-2018

Hi Guys, I am having trouble converting the ISO 8601 to a unixtimestamp . Here is what i have tried current_ts: 2018-11-11T00:17:27.937000 Using updateAttribute, I have configured the below property ${current_ts:toDate("yyyy-MM-dd'T'HH:mm:ss.SSSSSS"):toNumber()} This gives the Output value : 1541863984000 if converted back from epochcoverter, its giving Assuming that this timestamp is in milliseconds: GMT: Saturday, 10 November 2018 15:33:04 Your time zone: Sunday, 11 November 2018 00:33:04 GMT+09:00 Relative: 22 days ago I have also tried using ${current_ts:toDate("yyyy-MM-dd'T'HH:mm:ss.SSSSSS", "UTC"):toNumber()} but the result is the same. Kindly help finding out correct expression. Regards, Faisal

te04_0172 · ‎11-05-2018

Hi, We are consuming kafka messages from different consumer groups and then distributing it on the cluster using RPG. Each flow file takes a different route before finally getting to PutHbasejson processor. We need to make sure that they get inserted in Hbase depending on the order they were consumed e.g Update should only be inserted after Insert etc. But since they go through different routes they often reach out of order to hbase. I have tried using the timestamp feature to version rows in Hbase but often times both update/delete have the same time stamp (which is technically incorrect behavior). Anyway the only other option left is to make sure to enforce the order. I have tried understanding the enforce order processor but its a little confusing PFB 1.As I understand Enforce order will only order the flow files on a single node and not the whole cluster? If so,How do i merge all the kafka messages on a single node? 2. There is property called Group identifier in the processor, we have different consumer ids in the consume kafka processor but i think consumer id cannot be accessed as attribute so how do i ensure that it only enforces order from messages from a specific consumer? I would ideally want to version the rows in hbase based on the timestamp(maybe concatenated with Primary keys) but i haven't been able to find any example of that. Please share if you have worked with something similar.

te04_0172 · ‎10-11-2018

Does any one know the reason of this error on kafka [ReplicaManager broker=1002] Error processing append operation on partition TABLE-0 (kafka.server.ReplicaManager) org.apache.kafka.common.errors.UnknownProducerIdException: Found no record of producerId=96007 on the broker. It is possible that the last message with the producerId=96007 has been removed due to hitting the retention limit.

te04_0172 · ‎10-04-2018

Our golden gate server sends CDC data to Kafka using its kafka handler. kafka is configured to have multiple partition as its a multi broker cluster. However for some reason Golden gate is sending data only to a single partition of a topic instead of distributing it across each one of them in round robin fashion. I created a topic using the console and tried sending the messages but they were distributed evenly so there is something wrong with the golden gate configuration for kafka. PFAbootstrap.txt the boot strap server configuration and please suggest any changes that needs to be made. The image shows whats happening to partitions.

te04_0172 · ‎09-26-2018

Does any one know when Streams Messaging Manager will be available for download? I would like to use it on our HDF 3.2 cluster.

te04_0172 · ‎09-18-2018

I want to know what happens to the flow files which are in queue if Nifi runs to a OOM (java heap size) error. 1.Do they still remain in the queue when the instance recovers or after a restart? 2.Whats the best practices to follow if we are in this type of situation? We are consuming from Kafka topics so what we do is consume all the data again by changing the group id since we can't selectively consume data from Kafka based on timestamp.

te04_0172 · ‎07-26-2018

logserach-1.pnglogsearch-2.pngI am trying to see Nifi logs from log search service in HDF. As visible in the image the histogram shows a considerable amount of warn events but the service logs tab is empty. When i change the filter from 'last 1 hour' to 'last 24 hour' it does show the logs from yesterday but nothing from today. Please help.

te04_0172 · ‎07-11-2018

Hi @Shu, I was able to implement your idea of using MergerRecord -> PutHbaseJsonRecord using the Record reader controller services. However , I think there is a limitation in PutHbaseJsonRecord. We are syncing Oracle tables in Hbase using Goldengate and there are table with multiple PK's in oracle .We are creating a corresponding row key in Hbase by concatenating those PK's together. The PutHbaseJson allows to do that so we concatenate the PK's and pass it as an attribute to the processor. But the PutHbaseJsonRecord corresponding property is "Row Identifier Field Name" , so it is expecting the row key to be an element in the json that is read by the record reader. I've tried passing the same attribute i was sending to the PutHbaseJson but it doesn't work. Do you agree to this? I can think of a work around here where I transform the JSON to add the attribute(concatenated PK's) to it,which at this point i don't know how to do it . But even if i manage to do it then I will also need to change the schema as well. Kindly let me know if there is a better way to skin this cat.

te04_0172 · ‎07-05-2018

Hi, I've got two questions 1.We are using Remote Process Group with Raw transport protocol to distribute the data across four node cluster. I see the nifi app log has a lot of instance of the below error o.a.nifi.remote.SocketRemoteSiteListener Unable to communicate with remote instance Peer[url=nifi://xxx-xxxxxx.prod.xx.:59528] (SocketFlowFileServerProtocol[CommsID=0bf887ed-acb3-4eea-94ac-5abf53ad0bf1]) due to java.io.EOFException; closing connection These error do not show on the bulletin board and nor do I see any data loss. I was curious to know if there is some bad configuration that is causing this to happen. 2. The app log also has the below error o.a.n.r.c.socket.EndpointConnectionPool EndpointConnectionPool[Cluster URL=[http://xxx-xxxxxx.prod.xx.local:9090/nifi-api]] Peer[url=nifi://ins-btrananifi107z.prod.jp.local:5001] indicates that port 417e3d23-5b1a-1616-9728-9d9d1a462646's destination is full; penalizing peer The data flow consume a high volume data and there is back pressure on almost all the connections. So probably that is what causing it. I guess there isn't much we can do here and once the back pressure resolve ,the error goes away on its own.Please let me know of your view.

te04_0172 · ‎06-26-2018

I think i miss understood the prupose of the record reader. It looks clear to me now. Thank you for the suggestion. I'll work on this idea.

te04_0172 · ‎06-26-2018

Hi Shu, Thank you for your reply. I'll have to study about this Record reader stuff in detail because by the time our flow file reaches the puthbaseJson proc it already contains the reduced json in its payload and simply needs to be put in the target hbase table. So I don't require any sort of manipulation or parsing to be done on it which apparently Record readers does. And i see that the record reader is a required field so there is no way around it. Is there a way to create a dummy reader that does nothing ? : P . I'll explore this on my own as well.

te04_0172 · ‎06-25-2018

The slowest part of our data flow is the PuthbaseJson processor and I am trying to find out a way to optimize it. There is a configuration inside the proc where you can increase the batch size of the flow files it can process in a single execution. It set to 25 by default and i have tried increasing it up to 1000 with little performance gain. Increasing the concurrent tasks also hasn't helped in speeding up the put commands that processor runs. Has any one else worked with this processor and optimized it? The batch configuration of the processor says that it does the put by first grouping the flowfiles by table. Is there any thing i can do here? the name of the table already comes with the flowfile as an attribute and it is extracted from the attribute using the expression language. I am not sure how do i 'group it' before it reaches the Puthbasejosn. Kindly let me know of any ideas.

te04_0172 · ‎06-20-2018

How would a consumer with multiple topics work? Let say for example in a four node cluster we have a consumer (consume_kafka proc with 1 concurrent task) consuming from 10 topics (separated with comma) . Will it assign the 4 concurrent task to these 10 topics on a round-robin manner? I found that creating individual processor as opposed to combining all the topics in one processor to be more efficient. But this where we are stuck , we need to consume from 250 sources so please let me know what would be an efficient approach here. Creating 250 processors is possible but then due to limited number of threads available , some or alot processors does get the thread they need and it ends up with an error.

te04_0172 · ‎06-11-2018

@mrodriguez Thanks for your reply. Please find below the output. []$ bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic TEST_KAFKA_TOPIC Topic:TEST_KAFKA_TOPIC PartitionCount:1 ReplicationFactor:1 Configs: Topic: TEST_KAFKA_TOPIC Partition: 0 Leader: 1001 Replicas: 1001 Isr: 1001 Yes,I'm able to consume from the topic from the console . The error comes and go on its own.

te04_0172 · ‎06-08-2018

Does anyone know about this error from Kafka? I am using Nifi 1.5.0 with ConsumerKafka processor. ConsumeKafka[id=34753ed3-9dd6-15ed-9c91-147026236eee] Failed to retain connection due to No current assignment for partition TEST_KAFKA_TOPIC: This is the first time we are testing Nifi to consume from over 200 topics and its failing terribly so far. When this error goes the other one comes up which is as below Was interrupted while trying to communicate with Kafka with lease org.apache.nifi.processors.kafka.pubsub.ConsumerPool$SimpleConsumerLease@6cb8afba. Will roll back session and discard any partially received data.

te04_0172 · ‎06-08-2018

I need to consume golden gate messages from over 250 tables(topics) . I wanted to know what would be the best practice to do that? It appears that the consume kafka processor has an upper limit of the number of topics it can handle which is 99. I've tried using 1 processor to consume from 99 topics but it became very difficult to debug in case of any error and I had to put in a separate processor for that particular topic. Creating 250 consume kafka processor will just clutter the UI and wont be a good approach i guess. I wanted to know how every one else is getting around this scenario ?

te04_0172 · ‎06-07-2018

Is there a recommended way to ensure the row counts form tables in source (Oracle) are consistent with that of target tables in Hbase ( data-lake)? .We are using Nifi which receives the golden gate messages and then by using different processor we store the transactions in Hbase ,so essentially the tables in Hbase should be in sync with the tables in Oracle at all times. I am interested in knowing how the teams ensure and proof this ? Do they take row counts from source and target everyday and match it and say that its synced ? I used the counter option in Nifi which maintained the record received against each table but i guess that is not an optimized way to do it.

te04_0172 · ‎04-05-2018

Hi, Is there a way to consume a kafka message along with its timestamp in Nifi using the Consumekafka processor? e.g. we can consume the kafka message and see the time stamp by adding the property print.timestamp=true . The output looks something like this on the console CreateTime:1522893745217 test_message I can't seem to access this variable 'CreateTime'. I've tried using ${kafka.CreateTime} in the Updateattribute process but it doesnt work. Please let me know if there is way to do this as adding the custom timestamp (now()) is not an option in our case.

te04_0172 · ‎03-22-2018

Hi All, Has anyone worked with the PutHabseJSON processor Timestamp field? I'm trying to pass an attribute(op_ts) which contains the time stamp("2018-03-09 01:15:15.000000") as below Timestamp : ${op_ts} But it errors out with the below message Invalid timestamp value: 2018-03-09 01:15:15.000000: java.lang.NumberFormatException: For input string: "2018-03-09 01:15:15.000000" I then tried converting it into number format using the below EL Timestamp :${op_ts:toDate("YYYY-MM-dd HH:mm:ss.SSSSSS"):toNumber()} The insert works fine but the timestamp value in Hbase is not correct. E.g. "2018-03-10 05:15:15.000000" is converted into Hbase timestamp 1514664915000 ( when i convert it back to Timestamp using (https://www.epochconverter.com/) it turns out to be Sunday, December 31, 2017 5:15:15 AM "2017-06-03 05:15:15.000000" is converted into Hbase timestamp 1483215315000 ( when i convert it back to Timestamp using it turns out to be Sunday, January 1, 2017 5:15:15 AM Please help!

te04_0172 · ‎03-13-2018

@Matt Clarke Thanks for your answer. If you can kindly aslo help with the below ones as well 1. While assigning salves and clients during HDF installation is 1 node enough for installing the Nifi Certificate authority ( considering we will have Nifi master service on 4 nodes) 2. How many zoo keeper client/ Infra Solr client are enough for a 3 node zookeeper cluster ( i.e zoo keeper master service on 3 nodes)? and is it okay to have client service running on the same node which is also hosting the zookeeper master service or should it be on a different server? Thank you.

te04_0172 · ‎03-12-2018

Hi All. I am trying to install the HDF cluster on a 9 node cluster. Previously I have only worked on a single node Nifi standalone instance so its a kind big jump for me. So couple of question here 1. There is no concept of Slaves when talking about Nifi right? When 'Assigning masters' I need to add all nodes on which i need to install the Nifi service and that is how i will get a truly "9 node Nifi cluster"? By default the Ambari interface is recommending to install nifi only on 3 nodes. 2. By default Ambari is installing zookeeper on 3 nodes ( strangely these are the same nodes on which its installing Nifi) Why is it not using/recommending the rest of the nodes on which Ambari host is already installed. Do i have to install zookeeper client on every node i install nifi on? 3. In Nifi cluster is there single coordinator node where keeps the cluster together? e.g. My data flows puts all the flow files it recieves from golden gate into a folder at OS level. Do i have to create folder on every node to keep the cluster at sync?

te04_0172 · ‎02-20-2018

I did not make any changes to the consumer process or the group id. The offset reset option is set to latest. There were also no changes made to the topic on the consumer side at least. The topic which in this case is a name of an Oracle table is generated in kafka when the table is created in Oracle. I am pretty sure there were no changes made in the configuration on goldengate as well. As the team that manages those server have made changes to Goldengate configuration in the past with no disruption to workflows in Nifi. The server on which Goldengate is installed is different from the one that hosts Nifi , Can a network disruption cause this error?

te04_0172 · ‎02-16-2018

Hi All, I got an error while using the ConsumeKafka_0_10 processor . Below is the complete error from nifi-app.log 2018-02-16 11:47:11,465 WARN [Timer-Driven Process Thread-2] o.a.n.p.kafka.pubsub.ConsumeKafka_0_10 ConsumeKafka_0_10[id=b846165f-115a-1161-cadd-815bbb3a45dd] Was interrupted while trying to communicate with Kafka with lease org.apache.nifi.processors.kafka.pubsub.ConsumerPool$SimpleConsumerLease@6cb8afba. Will roll back session and discard any partially received data. bootstrap.servers = [xxx-xxxkafka-xxxz.xxx.xx.local:9093] key.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor] value.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer The last time I saw this error I restarted the nifi and it was gone but it did not work this time. There hasn't been any change on the kafka side. I'm using nifi 1.4.0 along with kafka_2.11-0.11.0.1 to consume kafka records , please let me know if any one knows the root cause of this error.

te04_0172 · ‎02-15-2018

Hi @spdvnz @Raj B , Can you let me know how were you able to grab the failures eventually? I've tried storing the bulletin board messages in hdfs using the REST API but the Json generated from there is very detailed and would require a lot of work before I can use it for monitoring purpose. Would like to know how you guys did it? Here is the link to the actual question https://community.hortonworks.com/questions/171150/monitoring-nifi-flow-file-failures-success-1.html

te04_0172 · ‎02-15-2018

Is there any way to increase this window from 5 minutes to lets say an hour or more? ( A common question would be how many records were processed during a 24 hour window etc.)

te04_0172 · ‎02-14-2018

Thanks, can you kindly let me know how can I change the retention period of these repositories? (from the nifi properties file I can see these two properties whose unit are the length of time. nifi.flow.configuration.archive.max.time=30 days nifi.content.repository.archive.max.retention.period=12 hours )

te04_0172 · ‎02-13-2018

I am working on building a monitoring solution for my Nifi workflow (Real time data lake using Golden-gate/ Kafka). Currently I am storing all the Goldgen gate records/flowfiles received from kafka in to a hdfs directory and at the end of the workflow ,in case its ingested successfully in hbase the flowfile is deleted from the directory. So i know that the json format flowfiles left in the Hdfs directory are ones that have failed. Now the issue with finding the reason of the failure is that Nifi's bulletin board shows only the record for last 5/10 mins (not sure of the duration) . I've tried storing the bulletin board messages in hdfs using the REST API but the Json generated from there is very detailed and would require a lot of work before I can use it for monitoring purpose. Has anyone else worked with such type of monitoring? I would also like to know the throughput of the workflow which would include the no of records failed or successfully ingested etc. I know i can get the last-5-min stat from the status history. But if any one else has worked on a similar monitoring task kindly let me know.

te04_0172 · ‎01-29-2018

Thanks, I was able to build the code and get the nar files.

te04_0172 · ‎01-16-2018

New feature request Jira for deleting a Hbase row using Nifi processor has been marked as 'Patch available' (https://issues.apache.org/jira/browse/NIFI-3538) . There is a github link for the pull request mentioned in the ticket as well. Can some one tell how to add this to our instance of Nifi so that we can access the processor? The github link is as below https://github.com/apache/nifi/pull/2294

te04_0172 · ‎01-16-2018

Hi, The Jira has been marked as "Patch available". Can you kindly let me know how can we 'install' or upgrade our instance of Nifi so that we can access this processor?

Online	Offline
Last Visited	‎02-25-2019 11:55 AM

Date Registered	‎01-09-2018 01:47 AM
Last Visited	‎02-25-2019 11:55 AM
Total Messages Posted	32
Total Kudos Received	3

Expression language - Convert ISO 8601 to unixtime...

Enforce ordering on kafka messages in nifi

Kafka error

Golden gate kafka integration issue

Streams Messaging Manager

Nifi heap size

HDF LogSearch not showing logs

Re: Puthbasejson performance optimization

RPG S2S Error

Re: Puthbasejson performance optimization

Re: Puthbasejson performance optimization

Puthbasejson performance optimization

Re: Integrating Apache NiFi and Apache Kafka

Re: Consume Kafka error Nifi

Consume Kafka error Nifi

Multiple topics using Nifi consume kafka

Row counts - date lake

Read Kafka timestamp as an attribute in Nifi

PutHBaseJSON - Timestamp

Re: HDF Ambari Nifi

HDF Ambari Nifi

Re: Nifi ConsumeKafka_0_10 error

Nifi ConsumeKafka_0_10 error

Re: Capture Error messages from NiFi bulletins

Re: Counters as Custom Processor Metrics

Re: Monitoring Nifi - flow file failures/ success

Monitoring Nifi - flow file failures/ success

Re: NiFi - Delete Hbase Row

Upgrading Nifi with the new patch

Re: NiFi - Delete Hbase Row