How to get kafka partition ID?

[Yuta Morisawa]

Hi

I want to apply a window function simply on data from each
kafka-partition(I mean I don't need any grouping and just want to
process data parallely).
I think the best way to do so is drawing kafka partition id and use it
for keyBy function.
For example,

val data = stream.keyBy(kafka-partitionID)
                  .window(...)
                  .fold(...)

But, I could not find the way.
How can I get the kafka-partition ID in Flink code?


--
Regards,
Yuta

[Gordon Weakliem]

If you pass a KeyedDeserializationSchema<T> to new FlinkKafkaConsumer08(topic, keyedDeserializationSchema, properties), you'll implement a method like this: 

    public T deserialize(byte[] messageKey, byte[] message, String topic, int partition, long offset) throws IOException {

    }

Then just make T a type that contains the partition as a field so operators down the pipeline can access that field.

On Tue, Jan 16, 2018 at 12:11 AM, Yuta Morisawa <[hidden email]> wrote:

Hi

I want to apply a window function simply on data from each kafka-partition(I mean I don't need any grouping and just want to process data parallely).
I think the best way to do so is drawing kafka partition id and use it for keyBy function.
For example,

val data = stream.keyBy(kafka-partitionID)
                 .window(...)
                 .fold(...)

But, I could not find the way.
How can I get the kafka-partition ID in Flink code?


--
Regards,
Yuta

--

[Yuta Morisawa]

Hi Gordon

Thanks a lot!
So far I used AbstractDeserializationSchema.
I will try the class you mentioned.

Regards
On 2018/01/17 2:48, Gordon Weakliem wrote:

> If you pass a KeyedDeserializationSchema<T> to
> new FlinkKafkaConsumer08(topic, keyedDeserializationSchema, properties),
> you'll implement a method like this:
>
>      public T deserialize(byte[] messageKey, byte[] message, String
> topic, int partition, long offset) throws IOException {
>      }
>
> Then just make T a type that contains the partition as a field so
> operators down the pipeline can access that field.
>
>
> On Tue, Jan 16, 2018 at 12:11 AM, Yuta Morisawa
> <[hidden email] <mailto:[hidden email]>> wrote:
>
>     Hi
>
>     I want to apply a window function simply on data from each
>     kafka-partition(I mean I don't need any grouping and just want to
>     process data parallely).
>     I think the best way to do so is drawing kafka partition id and use
>     it for keyBy function.
>     For example,
>
>     val data = stream.keyBy(kafka-partitionID)
>                       .window(...)
>                       .fold(...)
>
>     But, I could not find the way.
>     How can I get the kafka-partition ID in Flink code?
>
>
>     --
>     Regards,
>     Yuta
>
>
>
>
> --
> Img
> *  Gordon Weakliem*| Sr. Software Engineer
>
> *O *303.493.5490
> *  Boulder*|NYC | London
> <https://twitter.com/sovrnholdings><https://www.facebook.com/sovrnholdings/><https://www.linkedin.com/company/3594890/>
>
>
>
>
> CONFIDENTIALITY. This communication is intended only for the use of the
> intended recipient(s) and contains information that is privileged and
> confidential. As a recipient of this confidential and proprietary
> information, you are prohibited from distributing this information
> outside of sovrn. Further, if you are not the intended recipient, please
> note that any dissemination of this communication is prohibited. If you
> have received this communication in error, please erase all copies of
> the message, including all attachments, and please also notify the
> sender immediately. Thank you for your cooperation.