You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flume.apache.org by "Jason J. W. Williams" <ja...@gmail.com> on 2016/06/06 23:32:29 UTC
Kafka Sink random partition assignment
Hi,
New to flume and I'm trying to relay log messages received over netcat
source to Kafka sink.
Everything seems to be fine, except that Flume is acting like it IS
assigning a partition key to the produced messages though none is assigned.
I'd like the messages to be assigned to a random partition, so that
consumers are load balanced.
* Flume 1.6.0
* Kafka 0.9.0.1
Flume config:
https://gist.github.com/williamsjj/8ae025906955fbc4b5f990e162b75d7c
Kafka topic config: kafka-topics --zookeeper localhost/kafka --create
--topic activity.history --partitions 20 --replication-factor 1
Python consumer program:
https://gist.github.com/williamsjj/9e67287f0154816c3a733a39ad008437
Test program (publishes to Flume):
https://gist.github.com/williamsjj/1eb097a187a3edb17ec1a3913e47e58b
Flume agent listens on 3132tcp for connections, and publishes messages
received to the Kafka activity.history topic. I'm running two instances of
the Python consumer.
What happens however, is all logs messages get sent to a single Kafka
consumer...if I restart Flume (leave consumers running) and re-run the
test, all messages get published to the other consumer. So it feels like
Flume is assigning a permanent partition key even though one is not defined
(and should therefore be random).
Any advice is greatly appreciated.
-J
Re: Kafka Sink random partition assignment
Posted by Gonzalo Herreros <gh...@gmail.com>.
In your config, the name of the partitioner is missing a T ("Paritioner"),
you should be getting an exception and maybe the sink is reverting to
partition by key:
relay_agent.sinks.activity_kafka_sink.kafka_partitioner.class =
org.apache.kafka.clients.producer.internals.RandomParitioner
Gonzalo
On 7 June 2016 at 00:32, Jason J. W. Williams <ja...@gmail.com>
wrote:
> Hi,
>
> New to flume and I'm trying to relay log messages received over netcat
> source to Kafka sink.
>
> Everything seems to be fine, except that Flume is acting like it IS
> assigning a partition key to the produced messages though none is assigned.
> I'd like the messages to be assigned to a random partition, so that
> consumers are load balanced.
>
> * Flume 1.6.0
> * Kafka 0.9.0.1
>
> Flume config:
> https://gist.github.com/williamsjj/8ae025906955fbc4b5f990e162b75d7c
>
> Kafka topic config: kafka-topics --zookeeper localhost/kafka --create
> --topic activity.history --partitions 20 --replication-factor 1
>
> Python consumer program:
> https://gist.github.com/williamsjj/9e67287f0154816c3a733a39ad008437
>
> Test program (publishes to Flume):
> https://gist.github.com/williamsjj/1eb097a187a3edb17ec1a3913e47e58b
>
> Flume agent listens on 3132tcp for connections, and publishes messages
> received to the Kafka activity.history topic. I'm running two instances of
> the Python consumer.
>
> What happens however, is all logs messages get sent to a single Kafka
> consumer...if I restart Flume (leave consumers running) and re-run the
> test, all messages get published to the other consumer. So it feels like
> Flume is assigning a permanent partition key even though one is not defined
> (and should therefore be random).
>
> Any advice is greatly appreciated.
>
> -J
>
Re: Kafka Sink random partition assignment
Posted by Jason Williams <ja...@gmail.com>.
Hey Chris,
That's great. I'll look into that. Thank you very much!
-J
Sent via iPhone
> On Jun 7, 2016, at 11:36, Chris Horrocks <ch...@hor.rocks> wrote:
>
> You can implement sink groups load balance processor (https://flume.apache.org/FlumeUserGuide.html#flume-sink-processors) to spread the load over multiple producers to partly account for situation where you might have one kafka sink rotating between mutliple partitions, it's not an exact science though as you'd need multiple kafka sinks (n number more than you have partitions in the topic being written to) in order to make the balance of probabilities such that you were always writing to each partition from at least one sink.
>
> As with all hadoop technologies: depends on your use-case.
>
> --
> Chris Horrocks
>
> From: Jason J. W. Williams <ja...@gmail.com>
> Reply: user@flume.apache.org <us...@flume.apache.org>
> Date: 7 June 2016 at 19:03:59
> To: user@flume.apache.org <us...@flume.apache.org>
> Subject: Re: Kafka Sink random partition assignment
>
>> Thanks again Chris. I am curious why I see the round-robin behavior I expected when using kafka-console-producer to inject messages though.
>>
>> -J
>>
>>> On Tuesday, June 7, 2016, Chris Horrocks <ch...@hor.rocks> wrote:
>>> It's by design of Kafka (and by extension flume). The producers are designed to be many-to-one (producers to partitions) and as such picking a random partition every 10 minutes prevents separate producer instances from all randomly picking the same partition.
>>>
>>> --
>>> Chris Horrocks
>>>
>>> From: Jason Williams <ja...@gmail.com>
>>> Reply: user@flume.apache.org <us...@flume.apache.org>
>>> Date: 7 June 2016 at 09:43:34
>>> To: user@flume.apache.org <us...@flume.apache.org>
>>> Subject: Re: Kafka Sink random partition assignment
>>>
>>>> Hey Chris,
>>>>
>>>> Thanks for help!
>>>>
>>>> Is that a limitation of the Flume Kafka sink or Kafka itself? Because when I use another Kafka producer and publish without a key, it randomly moves among the partitions on every publish.
>>>>
>>>> -J
>>>>
>>>> Sent via iPhone
>>>>
>>>> On Jun 7, 2016, at 00:08, Chris Horrocks <ch...@hor.rocks> wrote:
>>>>
>>>>> The producers bind to random partitions and move every 10 minutes. If you leave it long enough you should see it in the producer flume agent logs, although there's nothing to stop it from "randomly" choosing the same partition twice. Annoyingly there's no concept of producer groups (yet) to ensure that producers apportion the available partitions between them as this would create a synchronisation issue between what should be entirely independent processes.
>>>>>
>>>>> --
>>>>> Chris Horrocks
>>>>>> On 7 June 2016 at 00:32:29, Jason J. W. Williams (jasonjwwilliams@gmail.com) wrote:
>>>>>>
>>>>>> Hi,
>>>>>>
>>>>>> New to flume and I'm trying to relay log messages received over netcat source to Kafka sink.
>>>>>>
>>>>>> Everything seems to be fine, except that Flume is acting like it IS assigning a partition key to the produced messages though none is assigned. I'd like the messages to be assigned to a random partition, so that consumers are load balanced.
>>>>>>
>>>>>> * Flume 1.6.0
>>>>>> * Kafka 0.9.0.1
>>>>>>
>>>>>> Flume config: https://gist.github.com/williamsjj/8ae025906955fbc4b5f990e162b75d7c
>>>>>>
>>>>>> Kafka topic config: kafka-topics --zookeeper localhost/kafka --create --topic activity.history --partitions 20 --replication-factor 1
>>>>>>
>>>>>> Python consumer program: https://gist.github.com/williamsjj/9e67287f0154816c3a733a39ad008437
>>>>>>
>>>>>> Test program (publishes to Flume): https://gist.github.com/williamsjj/1eb097a187a3edb17ec1a3913e47e58b
>>>>>>
>>>>>> Flume agent listens on 3132tcp for connections, and publishes messages received to the Kafka activity.history topic. I'm running two instances of the Python consumer.
>>>>>>
>>>>>> What happens however, is all logs messages get sent to a single Kafka consumer...if I restart Flume (leave consumers running) and re-run the test, all messages get published to the other consumer. So it feels like Flume is assigning a permanent partition key even though one is not defined (and should therefore be random).
>>>>>>
>>>>>> Any advice is greatly appreciated.
>>>>>>
>>>>>> -J
Re: Kafka Sink random partition assignment
Posted by Chris Horrocks <ch...@hor.rocks>.
You can implement sink groups load balance processor (https://flume.apache.org/FlumeUserGuide.html#flume-sink-processors) to spread the load over multiple producers to partly account for situation where you might have one kafka sink rotating between mutliple partitions, it's not an exact science though as you'd need multiple kafka sinks (n number more than you have partitions in the topic being written to) in order to make the balance of probabilities such that you were always writing to each partition from at least one sink.
As with all hadoop technologies: depends on your use-case.
--
Chris Horrocks
From: Jason J. W. Williams <ja...@gmail.com>
Reply: user@flume.apache.org <us...@flume.apache.org>
Date: 7 June 2016 at 19:03:59
To: user@flume.apache.org <us...@flume.apache.org>
Subject: Re: Kafka Sink random partition assignment
Thanks again Chris. I am curious why I see the round-robin behavior I expected when using kafka-console-producer to inject messages though.
-J
On Tuesday, June 7, 2016, Chris Horrocks <ch...@hor.rocks> wrote:
It's by design of Kafka (and by extension flume). The producers are designed to be many-to-one (producers to partitions) and as such picking a random partition every 10 minutes prevents separate producer instances from all randomly picking the same partition.
--
Chris Horrocks
From: Jason Williams <ja...@gmail.com>
Reply: user@flume.apache.org <us...@flume.apache.org>
Date: 7 June 2016 at 09:43:34
To: user@flume.apache.org <us...@flume.apache.org>
Subject: Re: Kafka Sink random partition assignment
Hey Chris,
Thanks for help!
Is that a limitation of the Flume Kafka sink or Kafka itself? Because when I use another Kafka producer and publish without a key, it randomly moves among the partitions on every publish.
-J
Sent via iPhone
On Jun 7, 2016, at 00:08, Chris Horrocks <ch...@hor.rocks> wrote:
The producers bind to random partitions and move every 10 minutes. If you leave it long enough you should see it in the producer flume agent logs, although there's nothing to stop it from "randomly" choosing the same partition twice. Annoyingly there's no concept of producer groups (yet) to ensure that producers apportion the available partitions between them as this would create a synchronisation issue between what should be entirely independent processes.
--
Chris Horrocks
On 7 June 2016 at 00:32:29, Jason J. W. Williams (jasonjwwilliams@gmail.com) wrote:
Hi,
New to flume and I'm trying to relay log messages received over netcat source to Kafka sink.
Everything seems to be fine, except that Flume is acting like it IS assigning a partition key to the produced messages though none is assigned. I'd like the messages to be assigned to a random partition, so that consumers are load balanced.
* Flume 1.6.0
* Kafka 0.9.0.1
Flume config: https://gist.github.com/williamsjj/8ae025906955fbc4b5f990e162b75d7c
Kafka topic config: kafka-topics --zookeeper localhost/kafka --create --topic activity.history --partitions 20 --replication-factor 1
Python consumer program: https://gist.github.com/williamsjj/9e67287f0154816c3a733a39ad008437
Test program (publishes to Flume): https://gist.github.com/williamsjj/1eb097a187a3edb17ec1a3913e47e58b
Flume agent listens on 3132tcp for connections, and publishes messages received to the Kafka activity.history topic. I'm running two instances of the Python consumer.
What happens however, is all logs messages get sent to a single Kafka consumer...if I restart Flume (leave consumers running) and re-run the test, all messages get published to the other consumer. So it feels like Flume is assigning a permanent partition key even though one is not defined (and should therefore be random).
Any advice is greatly appreciated.
-J
Re: Kafka Sink random partition assignment
Posted by "Jason J. W. Williams" <ja...@gmail.com>.
Thanks again Chris. I am curious why I see the round-robin behavior I
expected when using kafka-console-producer to inject messages though.
-J
On Tuesday, June 7, 2016, Chris Horrocks <ch...@hor.rocks> wrote:
> It's by design of Kafka (and by extension flume). The producers are
> designed to be many-to-one (producers to partitions) and as such picking a
> random partition every 10 minutes prevents separate producer instances from
> all randomly picking the same partition.
>
> --
> Chris Horrocks
>
> From: Jason Williams <ja...@gmail.com>
> <javascript:_e(%7B%7D,'cvml','jasonjwwilliams@gmail.com');>
> Reply: user@flume.apache.org
> <javascript:_e(%7B%7D,'cvml','user@flume.apache.org');>
> <us...@flume.apache.org>
> <javascript:_e(%7B%7D,'cvml','user@flume.apache.org');>
> Date: 7 June 2016 at 09:43:34
> To: user@flume.apache.org
> <javascript:_e(%7B%7D,'cvml','user@flume.apache.org');>
> <us...@flume.apache.org>
> <javascript:_e(%7B%7D,'cvml','user@flume.apache.org');>
> Subject: Re: Kafka Sink random partition assignment
>
> Hey Chris,
>
> Thanks for help!
>
> Is that a limitation of the Flume Kafka sink or Kafka itself? Because when
> I use another Kafka producer and publish without a key, it randomly moves
> among the partitions on every publish.
>
> -J
>
> Sent via iPhone
>
> On Jun 7, 2016, at 00:08, Chris Horrocks <chris@hor.rocks
> <javascript:_e(%7B%7D,'cvml','chris@hor.rocks');>> wrote:
>
> The producers bind to random partitions and move every 10 minutes. If you
> leave it long enough you should see it in the producer flume agent logs,
> although there's nothing to stop it from "randomly" choosing the same
> partition twice. Annoyingly there's no concept of producer groups (yet) to
> ensure that producers apportion the available partitions between them as
> this would create a synchronisation issue between what should be entirely
> independent processes.
>
> --
> Chris Horrocks
>
> On 7 June 2016 at 00:32:29, Jason J. W. Williams (
> jasonjwwilliams@gmail.com
> <javascript:_e(%7B%7D,'cvml','jasonjwwilliams@gmail.com');>) wrote:
>
>> Hi,
>>
>> New to flume and I'm trying to relay log messages received over netcat
>> source to Kafka sink.
>>
>> Everything seems to be fine, except that Flume is acting like it IS
>> assigning a partition key to the produced messages though none is assigned.
>> I'd like the messages to be assigned to a random partition, so that
>> consumers are load balanced.
>>
>> * Flume 1.6.0
>> * Kafka 0.9.0.1
>>
>> Flume config:
>> https://gist.github.com/williamsjj/8ae025906955fbc4b5f990e162b75d7c
>>
>> Kafka topic config: kafka-topics --zookeeper localhost/kafka --create
>> --topic activity.history --partitions 20 --replication-factor 1
>>
>> Python consumer program:
>> https://gist.github.com/williamsjj/9e67287f0154816c3a733a39ad008437
>>
>> Test program (publishes to Flume):
>> https://gist.github.com/williamsjj/1eb097a187a3edb17ec1a3913e47e58b
>>
>> Flume agent listens on 3132tcp for connections, and publishes messages
>> received to the Kafka activity.history topic. I'm running two instances of
>> the Python consumer.
>>
>> What happens however, is all logs messages get sent to a single Kafka
>> consumer...if I restart Flume (leave consumers running) and re-run the
>> test, all messages get published to the other consumer. So it feels like
>> Flume is assigning a permanent partition key even though one is not defined
>> (and should therefore be random).
>>
>> Any advice is greatly appreciated.
>>
>> -J
>>
>
Re: Kafka Sink random partition assignment
Posted by Chris Horrocks <ch...@hor.rocks>.
It's by design of Kafka (and by extension flume). The producers are designed to be many-to-one (producers to partitions) and as such picking a random partition every 10 minutes prevents separate producer instances from all randomly picking the same partition.
--
Chris Horrocks
From: Jason Williams <ja...@gmail.com>
Reply: user@flume.apache.org <us...@flume.apache.org>
Date: 7 June 2016 at 09:43:34
To: user@flume.apache.org <us...@flume.apache.org>
Subject: Re: Kafka Sink random partition assignment
Hey Chris,
Thanks for help!
Is that a limitation of the Flume Kafka sink or Kafka itself? Because when I use another Kafka producer and publish without a key, it randomly moves among the partitions on every publish.
-J
Sent via iPhone
On Jun 7, 2016, at 00:08, Chris Horrocks <ch...@hor.rocks> wrote:
The producers bind to random partitions and move every 10 minutes. If you leave it long enough you should see it in the producer flume agent logs, although there's nothing to stop it from "randomly" choosing the same partition twice. Annoyingly there's no concept of producer groups (yet) to ensure that producers apportion the available partitions between them as this would create a synchronisation issue between what should be entirely independent processes.
--
Chris Horrocks
On 7 June 2016 at 00:32:29, Jason J. W. Williams (jasonjwwilliams@gmail.com) wrote:
Hi,
New to flume and I'm trying to relay log messages received over netcat source to Kafka sink.
Everything seems to be fine, except that Flume is acting like it IS assigning a partition key to the produced messages though none is assigned. I'd like the messages to be assigned to a random partition, so that consumers are load balanced.
* Flume 1.6.0
* Kafka 0.9.0.1
Flume config: https://gist.github.com/williamsjj/8ae025906955fbc4b5f990e162b75d7c
Kafka topic config: kafka-topics --zookeeper localhost/kafka --create --topic activity.history --partitions 20 --replication-factor 1
Python consumer program: https://gist.github.com/williamsjj/9e67287f0154816c3a733a39ad008437
Test program (publishes to Flume): https://gist.github.com/williamsjj/1eb097a187a3edb17ec1a3913e47e58b
Flume agent listens on 3132tcp for connections, and publishes messages received to the Kafka activity.history topic. I'm running two instances of the Python consumer.
What happens however, is all logs messages get sent to a single Kafka consumer...if I restart Flume (leave consumers running) and re-run the test, all messages get published to the other consumer. So it feels like Flume is assigning a permanent partition key even though one is not defined (and should therefore be random).
Any advice is greatly appreciated.
-J
Re: Kafka Sink random partition assignment
Posted by Gonzalo Herreros <gh...@gmail.com>.
By default a Kafka producer will choose a random partition, however, I
believe the Kafka sink by default partitions of the message key, so if the
key is null it won't do a good job.
On 7 June 2016 at 09:43, Jason Williams <ja...@gmail.com> wrote:
> Hey Chris,
>
> Thanks for help!
>
> Is that a limitation of the Flume Kafka sink or Kafka itself? Because when
> I use another Kafka producer and publish without a key, it randomly moves
> among the partitions on every publish.
>
> -J
>
> Sent via iPhone
>
> On Jun 7, 2016, at 00:08, Chris Horrocks <ch...@hor.rocks> wrote:
>
> The producers bind to random partitions and move every 10 minutes. If you
> leave it long enough you should see it in the producer flume agent logs,
> although there's nothing to stop it from "randomly" choosing the same
> partition twice. Annoyingly there's no concept of producer groups (yet) to
> ensure that producers apportion the available partitions between them as
> this would create a synchronisation issue between what should be entirely
> independent processes.
>
> --
> Chris Horrocks
>
> On 7 June 2016 at 00:32:29, Jason J. W. Williams (
> jasonjwwilliams@gmail.com) wrote:
>
>> Hi,
>>
>> New to flume and I'm trying to relay log messages received over netcat
>> source to Kafka sink.
>>
>> Everything seems to be fine, except that Flume is acting like it IS
>> assigning a partition key to the produced messages though none is assigned.
>> I'd like the messages to be assigned to a random partition, so that
>> consumers are load balanced.
>>
>> * Flume 1.6.0
>> * Kafka 0.9.0.1
>>
>> Flume config:
>> https://gist.github.com/williamsjj/8ae025906955fbc4b5f990e162b75d7c
>>
>> Kafka topic config: kafka-topics --zookeeper localhost/kafka --create
>> --topic activity.history --partitions 20 --replication-factor 1
>>
>> Python consumer program:
>> https://gist.github.com/williamsjj/9e67287f0154816c3a733a39ad008437
>>
>> Test program (publishes to Flume):
>> https://gist.github.com/williamsjj/1eb097a187a3edb17ec1a3913e47e58b
>>
>> Flume agent listens on 3132tcp for connections, and publishes messages
>> received to the Kafka activity.history topic. I'm running two instances of
>> the Python consumer.
>>
>> What happens however, is all logs messages get sent to a single Kafka
>> consumer...if I restart Flume (leave consumers running) and re-run the
>> test, all messages get published to the other consumer. So it feels like
>> Flume is assigning a permanent partition key even though one is not defined
>> (and should therefore be random).
>>
>> Any advice is greatly appreciated.
>>
>> -J
>>
>
Re: Kafka Sink random partition assignment
Posted by Jason Williams <ja...@gmail.com>.
Hey Chris,
Thanks for help!
Is that a limitation of the Flume Kafka sink or Kafka itself? Because when I use another Kafka producer and publish without a key, it randomly moves among the partitions on every publish.
-J
Sent via iPhone
> On Jun 7, 2016, at 00:08, Chris Horrocks <ch...@hor.rocks> wrote:
>
> The producers bind to random partitions and move every 10 minutes. If you leave it long enough you should see it in the producer flume agent logs, although there's nothing to stop it from "randomly" choosing the same partition twice. Annoyingly there's no concept of producer groups (yet) to ensure that producers apportion the available partitions between them as this would create a synchronisation issue between what should be entirely independent processes.
>
> --
> Chris Horrocks
>> On 7 June 2016 at 00:32:29, Jason J. W. Williams (jasonjwwilliams@gmail.com) wrote:
>>
>> Hi,
>>
>> New to flume and I'm trying to relay log messages received over netcat source to Kafka sink.
>>
>> Everything seems to be fine, except that Flume is acting like it IS assigning a partition key to the produced messages though none is assigned. I'd like the messages to be assigned to a random partition, so that consumers are load balanced.
>>
>> * Flume 1.6.0
>> * Kafka 0.9.0.1
>>
>> Flume config: https://gist.github.com/williamsjj/8ae025906955fbc4b5f990e162b75d7c
>>
>> Kafka topic config: kafka-topics --zookeeper localhost/kafka --create --topic activity.history --partitions 20 --replication-factor 1
>>
>> Python consumer program: https://gist.github.com/williamsjj/9e67287f0154816c3a733a39ad008437
>>
>> Test program (publishes to Flume): https://gist.github.com/williamsjj/1eb097a187a3edb17ec1a3913e47e58b
>>
>> Flume agent listens on 3132tcp for connections, and publishes messages received to the Kafka activity.history topic. I'm running two instances of the Python consumer.
>>
>> What happens however, is all logs messages get sent to a single Kafka consumer...if I restart Flume (leave consumers running) and re-run the test, all messages get published to the other consumer. So it feels like Flume is assigning a permanent partition key even though one is not defined (and should therefore be random).
>>
>> Any advice is greatly appreciated.
>>
>> -J
Re: Kafka Sink random partition assignment
Posted by Chris Horrocks <ch...@hor.rocks>.
The producers bind to random partitions and move every 10 minutes. If you leave it long enough you should see it in the producer flume agent logs, although there's nothing to stop it from "randomly" choosing the same partition twice. Annoyingly there's no concept of producer groups (yet) to ensure that producers apportion the available partitions between them as this would create a synchronisation issue between what should be entirely independent processes.
--
Chris Horrocks
On 7 June 2016 at 00:32:29, Jason J. W. Williams (jasonjwwilliams@gmail.com(mailto:jasonjwwilliams@gmail.com)) wrote:
>
> Hi,
>
> New to flume and I'm trying to relay log messages received over netcat source to Kafka sink.
>
> Everything seems to be fine, except that Flume is acting like it IS assigning a partition key to the produced messages though none is assigned. I'd like the messages to be assigned to a random partition, so that consumers are load balanced.
>
> * Flume 1.6.0
> * Kafka 0.9.0.1
>
> Flume config: https://gist.github.com/williamsjj/8ae025906955fbc4b5f990e162b75d7c
>
> Kafka topic config: kafka-topics --zookeeper localhost/kafka --create --topic activity.history --partitions 20 --replication-factor 1
>
> Python consumer program: https://gist.github.com/williamsjj/9e67287f0154816c3a733a39ad008437
>
> Test program (publishes to Flume): https://gist.github.com/williamsjj/1eb097a187a3edb17ec1a3913e47e58b
>
> Flume agent listens on 3132tcp for connections, and publishes messages received to the Kafka activity.history topic. I'm running two instances of the Python consumer.
>
> What happens however, is all logs messages get sent to a single Kafka consumer...if I restart Flume (leave consumers running) and re-run the test, all messages get published to the other consumer. So it feels like Flume is assigning a permanent partition key even though one is not defined (and should therefore be random).
>
> Any advice is greatly appreciated.
>
> -J