Return-Path: <user-return-12531-archive-asf-public=cust-asf.ponee.io@flink.apache.org>
X-Original-To: archive-asf-public-internal@cust-asf2.ponee.io
Delivered-To: archive-asf-public-internal@cust-asf2.ponee.io
Received: from cust-asf.ponee.io (cust-asf.ponee.io [163.172.22.183])
	by cust-asf2.ponee.io (Postfix) with ESMTP id 5FC9C200C5A
	for <archive-asf-public-internal@cust-asf2.ponee.io>; Tue, 18 Apr 2017 16:16:38 +0200 (CEST)
Received: by cust-asf.ponee.io (Postfix)
	id 5E5DA160BA1; Tue, 18 Apr 2017 14:16:38 +0000 (UTC)
Delivered-To: archive-asf-public@cust-asf.ponee.io
Received: from mail.apache.org (hermes.apache.org [140.211.11.3])
	by cust-asf.ponee.io (Postfix) with SMTP id 803DB160B90
	for <archive-asf-public@cust-asf.ponee.io>; Tue, 18 Apr 2017 16:16:37 +0200 (CEST)
Received: (qmail 22372 invoked by uid 500); 18 Apr 2017 14:16:36 -0000
Mailing-List: contact user-help@flink.apache.org; run by ezmlm
Precedence: bulk
List-Help: <mailto:user-help@flink.apache.org>
List-Unsubscribe: <mailto:user-unsubscribe@flink.apache.org>
List-Post: <mailto:user@flink.apache.org>
List-Id: <user.flink.apache.org>
Delivered-To: mailing list user@flink.apache.org
Received: (qmail 22363 invoked by uid 99); 18 Apr 2017 14:16:36 -0000
Received: from mail-relay.apache.org (HELO mail-relay.apache.org) (140.211.11.15)
    by apache.org (qpsmtpd/0.29) with ESMTP; Tue, 18 Apr 2017 14:16:36 +0000
Received: from [192.168.178.53] (ipservice-092-214-155-234.092.214.pools.vodafone-ip.de [92.214.155.234])
	by mail-relay.apache.org (ASF Mail Server at mail-relay.apache.org) with ESMTPSA id E5D271A0372
	for <user@flink.apache.org>; Tue, 18 Apr 2017 14:16:35 +0000 (UTC)
Subject: Re: Key by Task number
To: user@flink.apache.org
References: <1831168830.3306870.1492522696605.ref@mail.yahoo.com>
 <1831168830.3306870.1492522696605@mail.yahoo.com>
 <CAJV0mRvu2OWHzeoJq9QT8XRJteAzt0kT8B7tRLE_ox9V8-jnEQ@mail.gmail.com>
From: Chesnay Schepler <chesnay@apache.org>
Message-ID: <57739b99-b068-0417-6d91-da0d09f534b0@apache.org>
Date: Tue, 18 Apr 2017 16:16:37 +0200
User-Agent: Mozilla/5.0 (Windows NT 10.0; WOW64; rv:52.0) Gecko/20100101
 Thunderbird/52.0.1
MIME-Version: 1.0
In-Reply-To: <CAJV0mRvu2OWHzeoJq9QT8XRJteAzt0kT8B7tRLE_ox9V8-jnEQ@mail.gmail.com>
Content-Type: multipart/alternative;
 boundary="------------97F6B949E523FA49028685D3"
Content-Language: en-US
archived-at: Tue, 18 Apr 2017 14:16:38 -0000

This is a multi-part message in MIME format.
--------------97F6B949E523FA49028685D3
Content-Type: text/plain; charset=utf-8; format=flowed
Content-Transfer-Encoding: 7bit

If the number of combinations between partition and schemaID is limited 
then the subtask index could actually improve the distribution of values.

In any case, the easiest way to do this is to add a RichMapFunction 
after the flatMap, or modify the flatMap, to also include the subtask index.
Typically this would be done by creating a Tuple2 containing the index 
and value.

On 18.04.2017 15:43, Kamil Dziublinski wrote:
> I am not sure if you really need a keyby, your load will be 
> distributed among your map function without it.  But could you explain 
> a bit what is your sink doing?
>
>
> As for setting parallelism on the consumer remember that you wont have 
> higher parallelism than number of partitions in your topic.
> If you have 240 partitions that's fine, but if you have less than 
> other subtasks will be idle. Only one task can read from one partition 
> in parallel.
>
> On Tue, Apr 18, 2017 at 3:38 PM Telco Phone <telco5@yahoo.com 
> <mailto:telco5@yahoo.com>> wrote:
>
>
>     I am trying to use the task number as a keyby value to help fan
>     out the work load reading from kafka.
>
>
>     Given:
>
>            DataStream<SchemaRecord> stream =
>                     env.addSource(new
>     FlinkKafkaConsumer010<SchemaRecord>("topicA", schema, properties)
>                   ).setParallelism(240).flatMap(new
>     SchemaRecordSplit()).setParallelism(240).
>                           name("TopicA splitter").keyBy("partition",
>     "keyByHelper", "schemaId");
>
>           stream.addSink(new CustomMaprFsSink()).name("TopicA
>     Sink").setParallelism(240);
>
>
>     In the DeserialClass I am trying to get to the
>
>     getRuntimeContext().getIndexOfThisSubtask();
>
>     Which is only avaliable in the RichSinkFunction
>
>
>
>     The above is partition (by hour) , schemaID (avro schemaId) and I
>     would like to add the task number so that all 240 readers /
>     writers have something to do.
>
>     Any ideas ?
>
>
>


--------------97F6B949E523FA49028685D3
Content-Type: text/html; charset=utf-8
Content-Transfer-Encoding: 8bit

<html>
  <head>
    <meta http-equiv="Content-Type" content="text/html; charset=utf-8">
  </head>
  <body text="#000000" bgcolor="#FFFFFF">
    <div class="moz-cite-prefix">If the number of combinations between
      partition and schemaID is limited then the subtask index could
      actually improve the distribution of values.<br>
      <br>
      In any case, the easiest way to do this is to add a
      RichMapFunction after the flatMap, or modify the flatMap, to also
      include the subtask index.<br>
      Typically this would be done by creating a Tuple2 containing the
      index and value.<br>
      <br>
      On 18.04.2017 15:43, Kamil Dziublinski wrote:<br>
    </div>
    <blockquote type="cite"
cite="mid:CAJV0mRvu2OWHzeoJq9QT8XRJteAzt0kT8B7tRLE_ox9V8-jnEQ@mail.gmail.com">
      <div dir="ltr">I am not sure if you really need a keyby, your load
        will be distributed among your map function without it.  But
        could you explain a bit what is your sink doing?
        <div><br>
        </div>
        <div><br>
        </div>
        <div>As for setting parallelism on the consumer remember that
          you wont have higher parallelism than number of partitions in
          your topic.</div>
        <div>If you have 240 partitions that's fine, but if you have
          less than other subtasks will be idle. Only one task can read
          from one partition in parallel.</div>
      </div>
      <br>
      <div class="gmail_quote">
        <div dir="ltr">On Tue, Apr 18, 2017 at 3:38 PM Telco Phone &lt;<a
            href="mailto:telco5@yahoo.com" moz-do-not-send="true">telco5@yahoo.com</a>&gt;
          wrote:<br>
        </div>
        <blockquote class="gmail_quote" style="margin:0 0 0
          .8ex;border-left:1px #ccc solid;padding-left:1ex">
          <div>
            <div
              style="color:#000;background-color:#fff;font-family:Helvetica
              Neue,Helvetica,Arial,Lucida
              Grande,sans-serif;font-size:16px">
              <div
                id="m_3276328799012739452yui_3_16_0_ym19_1_1492521769712_3184"><br>
              </div>
              <div
                id="m_3276328799012739452yui_3_16_0_ym19_1_1492521769712_3184">I
                am trying to use the task number as a keyby value to
                help fan out the work load reading from kafka.</div>
              <div
                id="m_3276328799012739452yui_3_16_0_ym19_1_1492521769712_3184"><br>
              </div>
              <div
                id="m_3276328799012739452yui_3_16_0_ym19_1_1492521769712_3184"><br>
              </div>
              <div
                id="m_3276328799012739452yui_3_16_0_ym19_1_1492521769712_3184">Given:</div>
              <div
                id="m_3276328799012739452yui_3_16_0_ym19_1_1492521769712_3184"><br>
              </div>
              <div
                id="m_3276328799012739452yui_3_16_0_ym19_1_1492521769712_3184">
                       DataStream&lt;SchemaRecord&gt; stream =</div>
              <div
                id="m_3276328799012739452yui_3_16_0_ym19_1_1492521769712_3389"
                dir="ltr">                env.addSource(new
                FlinkKafkaConsumer010&lt;SchemaRecord&gt;("topicA",
                schema, properties)</div>
              <div
                id="m_3276328799012739452yui_3_16_0_ym19_1_1492521769712_3390"> 
                              ).setParallelism(240).flatMap(new
                SchemaRecordSplit()).setParallelism(240).</div>
              <div
                id="m_3276328799012739452yui_3_16_0_ym19_1_1492521769712_3391"> 
                                      name("TopicA
                splitter").keyBy("partition", "keyByHelper",
                "schemaId");</div>
              <div
                id="m_3276328799012739452yui_3_16_0_ym19_1_1492521769712_3391"><br>
              </div>
              <div dir="ltr"
                id="m_3276328799012739452yui_3_16_0_ym19_1_1492521769712_3394"> 
                      stream.addSink(new
                CustomMaprFsSink()).name("TopicA
                Sink").setParallelism(240);</div>
              <div dir="ltr"
                id="m_3276328799012739452yui_3_16_0_ym19_1_1492521769712_3394"><br>
              </div>
              <div dir="ltr"
                id="m_3276328799012739452yui_3_16_0_ym19_1_1492521769712_3394"><br>
              </div>
              <div dir="ltr"
                id="m_3276328799012739452yui_3_16_0_ym19_1_1492521769712_3394">In
                the DeserialClass I am trying to get to the </div>
              <div dir="ltr"
                id="m_3276328799012739452yui_3_16_0_ym19_1_1492521769712_3394"><br>
              </div>
              <div dir="ltr"
                id="m_3276328799012739452yui_3_16_0_ym19_1_1492521769712_3394">getRuntimeContext().getIndexOfThisSubtask(); <br>
              </div>
              <div dir="ltr"
                id="m_3276328799012739452yui_3_16_0_ym19_1_1492521769712_3394"><br>
              </div>
              <div dir="ltr"
                id="m_3276328799012739452yui_3_16_0_ym19_1_1492521769712_3394">Which
                is only avaliable in the RichSinkFunction</div>
              <div dir="ltr"
                id="m_3276328799012739452yui_3_16_0_ym19_1_1492521769712_3394"><br>
              </div>
              <div dir="ltr"
                id="m_3276328799012739452yui_3_16_0_ym19_1_1492521769712_3394"><br>
              </div>
              <div dir="ltr"
                id="m_3276328799012739452yui_3_16_0_ym19_1_1492521769712_3394"><br>
              </div>
              <div dir="ltr"
                id="m_3276328799012739452yui_3_16_0_ym19_1_1492521769712_3394">The
                above is partition (by hour) , schemaID (avro schemaId)
                and I would like to add the task number so that all 240
                readers / writers have something to do.</div>
              <div dir="ltr"
                id="m_3276328799012739452yui_3_16_0_ym19_1_1492521769712_3394"><br>
              </div>
              <div dir="ltr"
                id="m_3276328799012739452yui_3_16_0_ym19_1_1492521769712_3394">Any
                ideas ?</div>
              <div dir="ltr"
                id="m_3276328799012739452yui_3_16_0_ym19_1_1492521769712_3394"><br>
              </div>
              <div dir="ltr"
                id="m_3276328799012739452yui_3_16_0_ym19_1_1492521769712_3394"><br>
              </div>
              <div
                id="m_3276328799012739452yui_3_16_0_ym19_1_1492521769712_3184"><br>
              </div>
            </div>
          </div>
        </blockquote>
      </div>
    </blockquote>
    <p><br>
    </p>
  </body>
</html>

--------------97F6B949E523FA49028685D3--