Re: Need to measure start and end time for a processor
To add on to Lee’s excellent points, if you are concerned about capturing the
total lifetime, you may be able to intercept the messages coming from the
console application and insert an attribute for timing before Kafka processes
those messages. You can then evaluate against that in NiFi. This will include
the Kafka batching time, as the provenance metrics are NiFi-only.
Andy LoPresto
alopre...@apache.org
alopresto.apa...@gmail.com
PGP Fingerprint: 70EC B3E5 98A6 5A3F D3C4 BACE 3C6E F65B 2F7D EF69
> On May 27, 2016, at 2:43 PM, Lee Laim wrote:
>
> Igor,
>
> Here are 3 methods which may help:
>
>
> 1, Processor Statistics - 'Average Task Duration' is one of the processor
> statistics you can use to determine, on average, how fast an individual
> process takes. Additionally, 'Average Lineage duration' might give you some
> insight across multiple processors and queue times. (right click on the
> processor)
>
> 2. The Data Provenance feature has a graphical representation of the
> progression of the data flow, with an interactive slider bar.
> (4th icon on the upper right side of the browser)
>
> 3. You could also use two Update Attribute processors to make a 'elapsed'
> attribute to profile a specific portion of your data flow:
>
> start_time = ${now():toNumber()}
>
> data flow of some duration
>
> elapsed_time = ${now():toNumber():minus(${start_time})}
>
>
> -Lee
>
>
>
>
> On Fri, May 27, 2016 at 2:21 PM, Igor Kravzov > wrote:
> It's probably a strange requirement.
> I want to measure how long will take for Kafka/NiFi to process a certain
> amount of messages.
> From one pint there is a console application which will send messages to
> Kafka in a loop.
> GetKafka will read messages in batches and Splittext will split these.
> Basically I want to see how long it took NiFi to process messages from Kafka.
> Any idea how to achieve this?
>
> Thanks in advance.
>
signature.asc
Description: Message signed with OpenPGP using GPGMail
Re: Need to measure start and end time for a processor
Igor,
Here are 3 methods which may help:
1, Processor Statistics - 'Average Task Duration' is one of the
processor statistics you can use to determine, on average, how fast an
individual process takes. Additionally, 'Average Lineage duration' might
give you some insight across multiple processors and queue times. (right
click on the processor)
2. The Data Provenance feature has a graphical representation of the
progression of the data flow, with an interactive slider bar.
(4th icon on the upper right side of the browser)
3. You could also use two Update Attribute processors to make a 'elapsed'
attribute to profile a specific portion of your data flow:
start_time = ${now():toNumber()}
data flow of some duration
elapsed_time = ${now():toNumber():minus(${start_time})}
-Lee
On Fri, May 27, 2016 at 2:21 PM, Igor Kravzov
wrote:
> It's probably a strange requirement.
> I want to measure how long will take for Kafka/NiFi to process a certain
> amount of messages.
> From one pint there is a console application which will send messages to
> Kafka in a loop.
> GetKafka will read messages in batches and Splittext will split these.
> Basically I want to see how long it took NiFi to process messages from
> Kafka.
> Any idea how to achieve this?
>
> Thanks in advance.
>
Re: Apache NIFI insert to MySQL help
Hi Sven, you need to give the path of the driver jar file in the controller(Database file Jar Url).. Plus put the jar file in the nifi lib folder and restart Nifi. Anuj On Fri, May 27, 2016 at 4:37 PM, Sven Davison wrote: > I’m trying to do some stuff and send it to my MySQL database. I’m getting > an exception (see below) but I’m fairly sure it’s just because I don’t have > the right string or file located where it’s supposed to go. I’m not sure > what to put in for the database connection URL or the driver class name.. I > assume that’s where the problem is though. > > > > > > ProcessException: …. Cannot load JDBC driver class ‘com.mysql.jdbc.Driver’ > > Screenshot of actual error: http://prntscr.com/b9889i > > > > Controller service setup: > > Screenshot of setup: http://prntscr.com/b988xe > > > > > > I’m not seeing a lot on google at the moment, but I’m still lookin! > > > > > > > > -Sven >
Apache NIFI insert to MySQL help
I’m trying to do some stuff and send it to my MySQL database. I’m getting an exception (see below) but I’m fairly sure it’s just because I don’t have the right string or file located where it’s supposed to go. I’m not sure what to put in for the database connection URL or the driver class name.. I assume that’s where the problem is though. ProcessException: …. Cannot load JDBC driver class ‘com.mysql.jdbc.Driver’ Screenshot of actual error: http://prntscr.com/b9889i Controller service setup: Screenshot of setup: http://prntscr.com/b988xe I’m not seeing a lot on google at the moment, but I’m still lookin! -Sven
Need to measure start and end time for a processor
It's probably a strange requirement. I want to measure how long will take for Kafka/NiFi to process a certain amount of messages. >From one pint there is a console application which will send messages to Kafka in a loop. GetKafka will read messages in batches and Splittext will split these. Basically I want to see how long it took NiFi to process messages from Kafka. Any idea how to achieve this? Thanks in advance.
Re: GetKafka exception
Not bothering. This is good info we should document On May 27, 2016 11:47 AM, "Igor Kravzov" wrote: > Guys, I found the problem. It was my fault. > Kafka broker configured as fully qualified domain name. > What happens is when you call zookeper it returns FQDN but my hosts file > was not updated with IPs and node names. > > Sorry guys for bothering you. > > On Thu, May 26, 2016 at 8:08 PM Andrew Psaltis > wrote: > >> Hi Igor, >> It looks from the stack trace that the consumer is having trouble finding >> the leader for topic "Test1". Can you execute on one of your brokers, >> perhaps the one with ip address 10.1.132.113 the following command: >> >> $KAFKA_HOME/bin/kafka-topics.sh --describe --zookeeper --topic >> Test1 >> >> and let me know what the output is. >> >> Also another test that you can do to take NiFi out of the picture is to >> use the console consumer. You can execute that with the following command, >> again from one of your Kafka brokers. >> >> $KAFKA_HOME/bin/kafka-console-consumer.sh --zookeeper --topic Test1 >> >> If you do not have a process writing any data into kafka, then you can >> add the following parameter to the above command "--from-beginning" >> >> >> Looking forward to hearing back from you. >> >> Thanks, >> Andrew >> >> >> On Thu, May 26, 2016 at 5:44 PM, Igor Kravzov >> wrote: >> >>> Thanks Oleg. I'll try. If you find more data please let me know.. >>> >>> On Thu, May 26, 2016 at 5:39 PM, Oleg Zhurakousky < >>> ozhurakou...@hortonworks.com> wrote: >>> Igor, you may want to go they Kafka mailing list. I've seen this several times, and all I know it has to do with broker I'll look in more details once back on line Oleg Sent from my iPhone On May 26, 2016, at 17:20, Igor Kravzov wrote: Why I am getting exception bellow 2016-05-26 17:13:36,776 INFO [d217bceb-d58b-4031-9143-26d8ddd15e40_BPIW01a-1464297163195-b78e95a5-leader-finder-thread] kafka.consumer.ConsumerFetcherManager [ConsumerFetcherManager-1464297163222] Added fetcher for partitions ArrayBuffer() 2016-05-26 17:13:36,981 INFO [d217bceb-d58b-4031-9143-26d8ddd15e40_BPIW01a-1464297163195-b78e95a5-leader-finder-thread] kafka.utils.VerifiableProperties Verifying properties 2016-05-26 17:13:36,981 INFO [d217bceb-d58b-4031-9143-26d8ddd15e40_BPIW01a-1464297163195-b78e95a5-leader-finder-thread] kafka.utils.VerifiableProperties Property client.id is overridden to NiFi-d217bceb-d58b-4031-9143-26d8ddd15e40 2016-05-26 17:13:36,981 INFO [d217bceb-d58b-4031-9143-26d8ddd15e40_BPIW01a-1464297163195-b78e95a5-leader-finder-thread] kafka.utils.VerifiableProperties Property metadata.broker.list is overridden to 10.1.132.113:6667 2016-05-26 17:13:36,981 INFO [d217bceb-d58b-4031-9143-26d8ddd15e40_BPIW01a-1464297163195-b78e95a5-leader-finder-thread] kafka.utils.VerifiableProperties Property request.timeout.ms is overridden to 3 2016-05-26 17:13:36,981 INFO [d217bceb-d58b-4031-9143-26d8ddd15e40_BPIW01a-1464297163195-b78e95a5-leader-finder-thread] kafka.client.ClientUtils$ Fetching metadata from broker id:1001,host:10.1.132.113,port:6667 with correlation id 256 for 1 topic(s) Set(Test1) 2016-05-26 17:13:36,981 INFO [d217bceb-d58b-4031-9143-26d8ddd15e40_BPIW01a-1464297163195-b78e95a5-leader-finder-thread] kafka.producer.SyncProducer Connected to 10.1.132.113:6667 for producing 2016-05-26 17:13:36,981 INFO [d217bceb-d58b-4031-9143-26d8ddd15e40_BPIW01a-1464297163195-b78e95a5-leader-finder-thread] kafka.producer.SyncProducer Disconnecting from 10.1.132.113:6667 2016-05-26 17:13:36,981 WARN [d217bceb-d58b-4031-9143-26d8ddd15e40_BPIW01a-1464297163195-b78e95a5-leader-finder-thread] kafka.client.ClientUtils$ Fetching topic metadata with correlation id 256 for topics [Set(Test1)] from broker [id:1001,host:10.1.132.113,port:6667] failed java.nio.channels.ClosedChannelException: null at kafka.network.BlockingChannel.send(BlockingChannel.scala:100) ~[kafka_2.10-0.8.2.2.jar:na] at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:73) ~[kafka_2.10-0.8.2.2.jar:na] at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:72) ~[kafka_2.10-0.8.2.2.jar:na] at kafka.producer.SyncProducer.send(SyncProducer.scala:113) ~[kafka_2.10-0.8.2.2.jar:na] at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:58) [kafka_2.10-0.8.2.2.jar:na] at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:93) [kafka_2.10-0.8.2.2.jar:na] at kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:66) [kafka_2.10-0.8.2.2.jar:na] at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:60) [kafka_2.10-0.8.2.2.jar:na] 2016-05-26
Re: GetKafka exception
Guys, I found the problem. It was my fault. Kafka broker configured as fully qualified domain name. What happens is when you call zookeper it returns FQDN but my hosts file was not updated with IPs and node names. Sorry guys for bothering you. On Thu, May 26, 2016 at 8:08 PM Andrew Psaltis wrote: > Hi Igor, > It looks from the stack trace that the consumer is having trouble finding > the leader for topic "Test1". Can you execute on one of your brokers, > perhaps the one with ip address 10.1.132.113 the following command: > > $KAFKA_HOME/bin/kafka-topics.sh --describe --zookeeper --topic Test1 > > and let me know what the output is. > > Also another test that you can do to take NiFi out of the picture is to > use the console consumer. You can execute that with the following command, > again from one of your Kafka brokers. > > $KAFKA_HOME/bin/kafka-console-consumer.sh --zookeeper --topic Test1 > > If you do not have a process writing any data into kafka, then you can add > the following parameter to the above command "--from-beginning" > > > Looking forward to hearing back from you. > > Thanks, > Andrew > > > On Thu, May 26, 2016 at 5:44 PM, Igor Kravzov > wrote: > >> Thanks Oleg. I'll try. If you find more data please let me know.. >> >> On Thu, May 26, 2016 at 5:39 PM, Oleg Zhurakousky < >> ozhurakou...@hortonworks.com> wrote: >> >>> Igor, you may want to go they Kafka mailing list. I've seen this several >>> times, and all I know it has to do with broker >>> I'll look in more details once back on line >>> >>> Oleg >>> >>> Sent from my iPhone >>> >>> On May 26, 2016, at 17:20, Igor Kravzov wrote: >>> >>> Why I am getting exception bellow >>> >>> >>> 2016-05-26 17:13:36,776 INFO >>> [d217bceb-d58b-4031-9143-26d8ddd15e40_BPIW01a-1464297163195-b78e95a5-leader-finder-thread] >>> kafka.consumer.ConsumerFetcherManager >>> [ConsumerFetcherManager-1464297163222] Added fetcher for partitions >>> ArrayBuffer() >>> 2016-05-26 17:13:36,981 INFO >>> [d217bceb-d58b-4031-9143-26d8ddd15e40_BPIW01a-1464297163195-b78e95a5-leader-finder-thread] >>> kafka.utils.VerifiableProperties Verifying properties >>> 2016-05-26 17:13:36,981 INFO >>> [d217bceb-d58b-4031-9143-26d8ddd15e40_BPIW01a-1464297163195-b78e95a5-leader-finder-thread] >>> kafka.utils.VerifiableProperties Property client.id is overridden to >>> NiFi-d217bceb-d58b-4031-9143-26d8ddd15e40 >>> 2016-05-26 17:13:36,981 INFO >>> [d217bceb-d58b-4031-9143-26d8ddd15e40_BPIW01a-1464297163195-b78e95a5-leader-finder-thread] >>> kafka.utils.VerifiableProperties Property metadata.broker.list is >>> overridden to 10.1.132.113:6667 >>> 2016-05-26 17:13:36,981 INFO >>> [d217bceb-d58b-4031-9143-26d8ddd15e40_BPIW01a-1464297163195-b78e95a5-leader-finder-thread] >>> kafka.utils.VerifiableProperties Property request.timeout.ms is >>> overridden to 3 >>> 2016-05-26 17:13:36,981 INFO >>> [d217bceb-d58b-4031-9143-26d8ddd15e40_BPIW01a-1464297163195-b78e95a5-leader-finder-thread] >>> kafka.client.ClientUtils$ Fetching metadata from broker >>> id:1001,host:10.1.132.113,port:6667 with correlation id 256 for 1 topic(s) >>> Set(Test1) >>> 2016-05-26 17:13:36,981 INFO >>> [d217bceb-d58b-4031-9143-26d8ddd15e40_BPIW01a-1464297163195-b78e95a5-leader-finder-thread] >>> kafka.producer.SyncProducer Connected to 10.1.132.113:6667 for producing >>> 2016-05-26 17:13:36,981 INFO >>> [d217bceb-d58b-4031-9143-26d8ddd15e40_BPIW01a-1464297163195-b78e95a5-leader-finder-thread] >>> kafka.producer.SyncProducer Disconnecting from 10.1.132.113:6667 >>> 2016-05-26 17:13:36,981 WARN >>> [d217bceb-d58b-4031-9143-26d8ddd15e40_BPIW01a-1464297163195-b78e95a5-leader-finder-thread] >>> kafka.client.ClientUtils$ Fetching topic metadata with correlation id 256 >>> for topics [Set(Test1)] from broker [id:1001,host:10.1.132.113,port:6667] >>> failed >>> java.nio.channels.ClosedChannelException: null >>> at kafka.network.BlockingChannel.send(BlockingChannel.scala:100) >>> ~[kafka_2.10-0.8.2.2.jar:na] >>> at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:73) >>> ~[kafka_2.10-0.8.2.2.jar:na] >>> at >>> kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:72) >>> ~[kafka_2.10-0.8.2.2.jar:na] >>> at kafka.producer.SyncProducer.send(SyncProducer.scala:113) >>> ~[kafka_2.10-0.8.2.2.jar:na] >>> at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:58) >>> [kafka_2.10-0.8.2.2.jar:na] >>> at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:93) >>> [kafka_2.10-0.8.2.2.jar:na] >>> at >>> kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:66) >>> [kafka_2.10-0.8.2.2.jar:na] >>> at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:60) >>> [kafka_2.10-0.8.2.2.jar:na] >>> 2016-05-26 17:13:36,981 INFO >>> [d217bceb-d58b-4031-9143-26d8ddd15e40_BPIW01a-1464297163195-b78e95a5-leader-finder-thread] >>> kafka.producer.SyncProducer Disconnecting from 10.1.132.113:6667 >>> 2016-05-26 17:13:36,981 WARN >>> [d217bceb-d58b-
Cluster Node Protocol Threads - will this configuration help reduce a high queue?
I’m new to NiFi and I have a small problem. I’m running NiFi 0.7.0-SNAPSHOT on Windows 2012 R2 VMs. I’ve noticed that the queues on some of my Processors are very large, sometimes as high as 10,000 flowfiles. The flowfiles are eventually processed but at a slow pace. I run a cluster with 4 nodes. The initial Get Processor runs on the Primary Node to only request a single file and I then use the DistributeLoad Processor with the Next Available setting to spread the load across the cluster nodes. However, I see that the queue is highest on the Primary Node and the cluster nodes see very little work. My question is: will I increase throughput to the cluster nodes if I increase the nifi.cluster.node.protocol.threads from 2 to something higher? What effect does nifi.cluster.node.protocol.threads have on the nodes? Thanks, Kevin
Re: GetKafka exception
Hi ANdrew,
here is output for the first command
Topic:Test1 PartitionCount:1ReplicationFactor:1 Configs:
Topic: Test1Partition: 0Leader: 1001Replicas: 1001
Isr: 1001
Here is for second. I think i read the message.
{metadata.broker.list=xxxnode04.x.net:6667, request.timeout.ms=3,
client.id=console-consumer-85156, security.protocol=PLAINTEXT}
▒urlXhttp://
www.jpost.com/Middle-East/Analysis-What-Exactly-is-Happening-Saudi-Arabia-454018pageTypenews
addDate▒▒V▒TpostDate▒|.▒Tposttext
ghdhjdshd
PuTTY
^CProcessed a total of 1 messages
On Thu, May 26, 2016 at 8:08 PM, Andrew Psaltis
wrote:
> Hi Igor,
> It looks from the stack trace that the consumer is having trouble finding
> the leader for topic "Test1". Can you execute on one of your brokers,
> perhaps the one with ip address 10.1.132.113 the following command:
>
> $KAFKA_HOME/bin/kafka-topics.sh --describe --zookeeper --topic Test1
>
> and let me know what the output is.
>
> Also another test that you can do to take NiFi out of the picture is to
> use the console consumer. You can execute that with the following command,
> again from one of your Kafka brokers.
>
> $KAFKA_HOME/bin/kafka-console-consumer.sh --zookeeper --topic Test1
>
> If you do not have a process writing any data into kafka, then you can add
> the following parameter to the above command "--from-beginning"
>
>
> Looking forward to hearing back from you.
>
> Thanks,
> Andrew
>
>
> On Thu, May 26, 2016 at 5:44 PM, Igor Kravzov
> wrote:
>
>> Thanks Oleg. I'll try. If you find more data please let me know..
>>
>> On Thu, May 26, 2016 at 5:39 PM, Oleg Zhurakousky <
>> ozhurakou...@hortonworks.com> wrote:
>>
>>> Igor, you may want to go they Kafka mailing list. I've seen this several
>>> times, and all I know it has to do with broker
>>> I'll look in more details once back on line
>>>
>>> Oleg
>>>
>>> Sent from my iPhone
>>>
>>> On May 26, 2016, at 17:20, Igor Kravzov wrote:
>>>
>>> Why I am getting exception bellow
>>>
>>>
>>> 2016-05-26 17:13:36,776 INFO
>>> [d217bceb-d58b-4031-9143-26d8ddd15e40_BPIW01a-1464297163195-b78e95a5-leader-finder-thread]
>>> kafka.consumer.ConsumerFetcherManager
>>> [ConsumerFetcherManager-1464297163222] Added fetcher for partitions
>>> ArrayBuffer()
>>> 2016-05-26 17:13:36,981 INFO
>>> [d217bceb-d58b-4031-9143-26d8ddd15e40_BPIW01a-1464297163195-b78e95a5-leader-finder-thread]
>>> kafka.utils.VerifiableProperties Verifying properties
>>> 2016-05-26 17:13:36,981 INFO
>>> [d217bceb-d58b-4031-9143-26d8ddd15e40_BPIW01a-1464297163195-b78e95a5-leader-finder-thread]
>>> kafka.utils.VerifiableProperties Property client.id is overridden to
>>> NiFi-d217bceb-d58b-4031-9143-26d8ddd15e40
>>> 2016-05-26 17:13:36,981 INFO
>>> [d217bceb-d58b-4031-9143-26d8ddd15e40_BPIW01a-1464297163195-b78e95a5-leader-finder-thread]
>>> kafka.utils.VerifiableProperties Property metadata.broker.list is
>>> overridden to 10.1.132.113:6667
>>> 2016-05-26 17:13:36,981 INFO
>>> [d217bceb-d58b-4031-9143-26d8ddd15e40_BPIW01a-1464297163195-b78e95a5-leader-finder-thread]
>>> kafka.utils.VerifiableProperties Property request.timeout.ms is
>>> overridden to 3
>>> 2016-05-26 17:13:36,981 INFO
>>> [d217bceb-d58b-4031-9143-26d8ddd15e40_BPIW01a-1464297163195-b78e95a5-leader-finder-thread]
>>> kafka.client.ClientUtils$ Fetching metadata from broker
>>> id:1001,host:10.1.132.113,port:6667 with correlation id 256 for 1 topic(s)
>>> Set(Test1)
>>> 2016-05-26 17:13:36,981 INFO
>>> [d217bceb-d58b-4031-9143-26d8ddd15e40_BPIW01a-1464297163195-b78e95a5-leader-finder-thread]
>>> kafka.producer.SyncProducer Connected to 10.1.132.113:6667 for producing
>>> 2016-05-26 17:13:36,981 INFO
>>> [d217bceb-d58b-4031-9143-26d8ddd15e40_BPIW01a-1464297163195-b78e95a5-leader-finder-thread]
>>> kafka.producer.SyncProducer Disconnecting from 10.1.132.113:6667
>>> 2016-05-26 17:13:36,981 WARN
>>> [d217bceb-d58b-4031-9143-26d8ddd15e40_BPIW01a-1464297163195-b78e95a5-leader-finder-thread]
>>> kafka.client.ClientUtils$ Fetching topic metadata with correlation id 256
>>> for topics [Set(Test1)] from broker [id:1001,host:10.1.132.113,port:6667]
>>> failed
>>> java.nio.channels.ClosedChannelException: null
>>> at kafka.network.BlockingChannel.send(BlockingChannel.scala:100)
>>> ~[kafka_2.10-0.8.2.2.jar:na]
>>> at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:73)
>>> ~[kafka_2.10-0.8.2.2.jar:na]
>>> at
>>> kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:72)
>>> ~[kafka_2.10-0.8.2.2.jar:na]
>>> at kafka.producer.SyncProducer.send(SyncProducer.scala:113)
>>> ~[kafka_2.10-0.8.2.2.jar:na]
>>> at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:58)
>>> [kafka_2.10-0.8.2.2.jar:na]
>>> at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:93)
>>> [kafka_2.10-0.8.2.2.jar:na]
>>> at
>>> kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:66)
>>> [kafka_2.10-0.8.2.2.j
