From user-return-1312-archive-asf-public=cust-asf.ponee.io@kudu.apache.org Fri Mar 16 22:57:10 2018 Return-Path: X-Original-To: archive-asf-public@cust-asf.ponee.io Delivered-To: archive-asf-public@cust-asf.ponee.io Received: from mail.apache.org (hermes.apache.org [140.211.11.3]) by mx-eu-01.ponee.io (Postfix) with SMTP id 0D1E7180608 for ; Fri, 16 Mar 2018 22:57:08 +0100 (CET) Received: (qmail 1875 invoked by uid 500); 16 Mar 2018 21:57:08 -0000 Mailing-List: contact user-help@kudu.apache.org; run by ezmlm Precedence: bulk List-Help: List-Unsubscribe: List-Post: List-Id: Reply-To: user@kudu.apache.org Delivered-To: mailing list user@kudu.apache.org Received: (qmail 1859 invoked by uid 99); 16 Mar 2018 21:57:07 -0000 Received: from pnap-us-west-generic-nat.apache.org (HELO spamd2-us-west.apache.org) (209.188.14.142) by apache.org (qpsmtpd/0.29) with ESMTP; Fri, 16 Mar 2018 21:57:07 +0000 Received: from localhost (localhost [127.0.0.1]) by spamd2-us-west.apache.org (ASF Mail Server at spamd2-us-west.apache.org) with ESMTP id EED3D1A087A for ; Fri, 16 Mar 2018 21:57:06 +0000 (UTC) X-Virus-Scanned: Debian amavisd-new at spamd2-us-west.apache.org X-Spam-Flag: NO X-Spam-Score: 1.898 X-Spam-Level: * X-Spam-Status: No, score=1.898 tagged_above=-999 required=6.31 tests=[DKIM_SIGNED=0.1, DKIM_VALID=-0.1, DKIM_VALID_AU=-0.1, HTML_MESSAGE=2, RCVD_IN_DNSWL_NONE=-0.0001, RCVD_IN_MSPIKE_H2=-0.001, SPF_PASS=-0.001] autolearn=disabled Authentication-Results: spamd2-us-west.apache.org (amavisd-new); dkim=pass (1024-bit key) header.d=impactradius.com Received: from mx1-lw-eu.apache.org ([10.40.0.8]) by localhost (spamd2-us-west.apache.org [10.40.0.9]) (amavisd-new, port 10024) with ESMTP id g-MDH416HJYN for ; Fri, 16 Mar 2018 21:57:02 +0000 (UTC) Received: from mail-wr0-f171.google.com (mail-wr0-f171.google.com [209.85.128.171]) by mx1-lw-eu.apache.org (ASF Mail Server at mx1-lw-eu.apache.org) with ESMTPS id 50B015F2EC for ; Fri, 16 Mar 2018 21:57:02 +0000 (UTC) Received: by mail-wr0-f171.google.com with SMTP id l8so12981084wrg.5 for ; Fri, 16 Mar 2018 14:57:02 -0700 (PDT) DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=impactradius.com; s=google; h=mime-version:in-reply-to:references:from:date:message-id:subject:to; bh=98camESX3l2TYYyHSRhBDLh4lmh9Dq28B5FlfT7hRPk=; b=CtF47MwOitqStHTSphWmEOioy7A8CMNOqL1ch2W0pwbTY0MmCeyi4I36DuDxwOsezv yGXRvvuYoMtNAIB+GjbEtT1/bL3ma+z0hlm8L3nW/gN/0PfAo3f3HMw2WYPz4U0YlqZr +f5otr3fc47VDOiWP4r74VVyU+VyQo6tI0RM8= X-Google-DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=1e100.net; s=20161025; h=x-gm-message-state:mime-version:in-reply-to:references:from:date :message-id:subject:to; bh=98camESX3l2TYYyHSRhBDLh4lmh9Dq28B5FlfT7hRPk=; b=lMpBlq3EIqjOfR5DyyIccZ8xDilcE2HWVJF3QbeShGZnPFEZoNSXVVb9NoThRf8Bp9 3n8r/HmC1mpFlRREHHrHjpgpTPY+EJxECo+MyTAoIQr7AD6FqnHlYiidC5P6/F1gR1C7 CzJ6Z6tJ7cC+ZMAtsYXhbW/y2ViDu2LOVfi6+WlZTRzX+jwkRHnhRN3So0cAGeREOr8c K4s72cGJDSjQTTocbfiQAA1DzT/f5zahFg0pqhNWzL8orGE/N6QspigLuYKHm10UNw+q JHQvzwzBrZ3HrVi1PH7d5E4kump9/0xixaUcQz5mBgFHaq3w9CIyuFUZtCrekIh5LDbV zmjw== X-Gm-Message-State: AElRT7HjYNTCb4AioSDMg7Ed2jmJ+7QcSCfc1O2l4qcSbGdIp8bCNlcO PzPaukDVF0VcC1ZLfIkilmjj7ubh1B3V0bRqTOcX0dupfOE= X-Google-Smtp-Source: AG47ELsREc/oQhB+4oII3rTN14rDpxmp1JyyoEiQfFn05C1fLu2IJekXnkRDMRk2XrTmizmkyBunkZFG16yRxhKBbn0= X-Received: by 10.223.162.203 with SMTP id t11mr3061916wra.88.1521237415123; Fri, 16 Mar 2018 14:56:55 -0700 (PDT) MIME-Version: 1.0 Received: by 10.223.175.120 with HTTP; Fri, 16 Mar 2018 14:56:54 -0700 (PDT) In-Reply-To: References: <1BEF1E38-2AA2-4CF0-B9B5-CD138C166E80@mediamath.com> From: Mauricio Aristizabal Date: Fri, 16 Mar 2018 14:56:54 -0700 Message-ID: Subject: Re: "broadcast" tablet replication for kudu? To: user@kudu.apache.org Content-Type: multipart/alternative; boundary="f403045ea6c4bf74c305678eaf81" --f403045ea6c4bf74c305678eaf81 Content-Type: text/plain; charset="UTF-8" @Todd I know working with parquet in the past I've seen small dimensions that fit in 1 single file/block limit parallelism of join/exchange/aggregation nodes, and I've forced those dims to spread across 20 or so blocks by leveraging SET PARQUET_FILE_SIZE=8m; or similar when doing INSERT OVERWRITE to load them, which then allows these operations to parallelize across that many nodes. Wouldn't it be useful here for Cliff's small dims to be partitioned into a couple tablets to similarly improve parallelism? -m On Fri, Mar 16, 2018 at 2:29 PM, Todd Lipcon wrote: > On Fri, Mar 16, 2018 at 2:19 PM, Cliff Resnick wrote: > >> Hey Todd, >> >> Thanks for that explanation, as well as all the great work you're doing >> -- it's much appreciated! I just have one last follow-up question. Reading >> about BROADCAST operations (Kudu, Spark, Flink, etc. ) it seems the smaller >> table is always copied in its entirety BEFORE the predicate is evaluated. >> > > That's not quite true. If you have a predicate on a joined column, or on > one of the columns in the joined table, it will be pushed down to the > "scan" operator, which happens before the "exchange". In addition, there is > a feature called "runtime filters" that can push dynamically-generated > filters from one side of the exchange to the other. > > >> But since the Kudu client provides a serialized scanner as part of the >> ScanToken API, why wouldn't Impala use that instead if it knows that the >> table is Kudu and the query has any type of predicate? Perhaps if I >> hash-partition the table I could maybe force this (because that complicates >> a BROADCAST)? I guess this is really a question for Impala but perhaps >> there is a more basic reason. >> > > Impala could definitely be smarter, just a matter of programming > Kudu-specific join strategies into the optimizer. Today, the optimizer > isn't aware of the unique properties of Kudu scans vs other storage > mechanisms. > > -Todd > > >> >> -Cliff >> >> On Fri, Mar 16, 2018 at 4:10 PM, Todd Lipcon wrote: >> >>> On Fri, Mar 16, 2018 at 12:30 PM, Clifford Resnick < >>> cresnick@mediamath.com> wrote: >>> >>>> I thought I had read that the Kudu client can configure a scan for >>>> CLOSEST_REPLICA and assumed this was a way to take advantage of data >>>> collocation. >>>> >>> >>> Yea, when a client uses CLOSEST_REPLICA it will read a local one if >>> available. However, that doesn't influence the higher level operation of >>> the Impala (or Spark) planner. The planner isn't aware of the replication >>> policy, so it will use one of the existing supported JOIN strategies. Given >>> statistics, it will choose to broadcast the small table, which means that >>> it will create a plan that looks like: >>> >>> >>> +-------------------------+ >>> | | >>> +---------->build JOIN | >>> | | | >>> | | probe | >>> +--------------+ +-------------------------+ >>> | | | >>> | Exchange | | >>> +----+ (broadcast | | >>> | | | | >>> | +--------------+ | >>> | | >>> +---------+ | >>> | | +-----------------------+ >>> | SCAN | | | >>> | KUDU | | SCAN (other side) | >>> | | | | >>> +---------+ +-----------------------+ >>> >>> (hopefully the ASCII art comes through) >>> >>> In other words, the "scan kudu" operator scans the table once, and then >>> replicates the results of that scan into the JOIN operator. The "scan kudu" >>> operator of course will read its local copy, but it will still go through >>> the exchange process. >>> >>> For the use case you're talking about, where the join is just looking up >>> a single row by PK in a dimension table, ideally we'd be using an >>> altogether different join strategy such as nested-loop join, with the inner >>> "loop" actually being a Kudu PK lookup, but that strategy isn't implemented >>> by Impala. >>> >>> -Todd >>> >>> >>> >>>> If this exists then how far out of context is my understanding of it? >>>> Reading about HDFS cache replication, I do know that Impala will choose a >>>> random replica there to more evenly distribute load. But especially >>>> compared to Kudu upsert, managing mutable data using Parquet is painful. >>>> So, perhaps to sum thing up, if nearly 100% of my metadata scan are single >>>> Primary Key lookups followed by a tiny broadcast then am I really just >>>> splitting hairs performance-wise between Kudu and HDFS-cached parquet? >>>> >>>> From: Todd Lipcon >>>> Reply-To: "user@kudu.apache.org" >>>> Date: Friday, March 16, 2018 at 2:51 PM >>>> >>>> To: "user@kudu.apache.org" >>>> Subject: Re: "broadcast" tablet replication for kudu? >>>> >>>> It's worth noting that, even if your table is replicated, Impala's >>>> planner is unaware of this fact and it will give the same plan regardless. >>>> That is to say, rather than every node scanning its local copy, instead a >>>> single node will perform the whole scan (assuming it's a small table) and >>>> broadcast it from there within the scope of a single query. So, I don't >>>> think you'll see any performance improvements on Impala queries by >>>> attempting something like an extremely high replication count. >>>> >>>> I could see bumping the replication count to 5 for these tables since >>>> the extra storage cost is low and it will ensure higher availability of the >>>> important central tables, but I'd be surprised if there is any measurable >>>> perf impact. >>>> >>>> -Todd >>>> >>>> On Fri, Mar 16, 2018 at 11:35 AM, Clifford Resnick < >>>> cresnick@mediamath.com> wrote: >>>> >>>>> Thanks for that, glad I was wrong there! Aside from replication >>>>> considerations, is it also recommended the number of tablet servers be odd? >>>>> >>>>> I will check forums as you suggested, but from what I read after >>>>> searching is that Impala relies on user configured caching strategies using >>>>> HDFS cache. The workload for these tables is very light write, maybe a >>>>> dozen or so records per hour across 6 or 7 tables. The size of the tables >>>>> ranges from thousands to low millions of rows so so sub-partitioning would >>>>> not be required. So perhaps this is not a typical use-case but I think it >>>>> could work quite well with kudu. >>>>> >>>>> From: Dan Burkert >>>>> Reply-To: "user@kudu.apache.org" >>>>> Date: Friday, March 16, 2018 at 2:09 PM >>>>> To: "user@kudu.apache.org" >>>>> Subject: Re: "broadcast" tablet replication for kudu? >>>>> >>>>> The replication count is the number of tablet servers which Kudu will >>>>> host copies on. So if you set the replication level to 5, Kudu will put >>>>> the data on 5 separate tablet servers. There's no built-in broadcast table >>>>> feature; upping the replication factor is the closest thing. A couple of >>>>> things to keep in mind: >>>>> >>>>> - Always use an odd replication count. This is important due to how >>>>> the Raft algorithm works. Recent versions of Kudu won't even let you >>>>> specify an even number without flipping some flags. >>>>> - We don't test much much beyond 5 replicas. It *should* work, but >>>>> you may run in to issues since it's a relatively rare configuration. With >>>>> a heavy write workload and many replicas you are even more likely to >>>>> encounter issues. >>>>> >>>>> It's also worth checking in an Impala forum whether it has features >>>>> that make joins against small broadcast tables better? Perhaps Impala can >>>>> cache small tables locally when doing joins. >>>>> >>>>> - Dan >>>>> >>>>> On Fri, Mar 16, 2018 at 10:55 AM, Clifford Resnick < >>>>> cresnick@mediamath.com> wrote: >>>>> >>>>>> The problem is, AFIK, that replication count is not necessarily the >>>>>> distribution count, so you can't guarantee all tablet servers will have a >>>>>> copy. >>>>>> >>>>>> On Mar 16, 2018 1:41 PM, Boris Tyukin wrote: >>>>>> I'm new to Kudu but we are also going to use Impala mostly with Kudu. >>>>>> We have a few tables that are small but used a lot. My plan is replicate >>>>>> them more than 3 times. When you create a kudu table, you can specify >>>>>> number of replicated copies (3 by default) and I guess you can put there a >>>>>> number, corresponding to your node count in cluster. The downside, you >>>>>> cannot change that number unless you recreate a table. >>>>>> >>>>>> On Fri, Mar 16, 2018 at 10:42 AM, Cliff Resnick >>>>>> wrote: >>>>>> >>>>>>> We will soon be moving our analytics from AWS Redshift to >>>>>>> Impala/Kudu. One Redshift feature that we will miss is its ALL >>>>>>> Distribution, where a copy of a table is maintained on each server. We >>>>>>> define a number of metadata tables this way since they are used in nearly >>>>>>> every query. We are considering using parquet in HDFS cache for these, and >>>>>>> Kudu would be a much better fit for the update semantics but we are worried >>>>>>> about the additional contention. I'm wondering if having a Broadcast, or >>>>>>> ALL, tablet replication might be an easy feature to add to Kudu? >>>>>>> >>>>>>> -Cliff >>>>>>> >>>>>> >>>>>> >>>>> >>>> >>>> >>>> -- >>>> Todd Lipcon >>>> Software Engineer, Cloudera >>>> >>>> >>> >>> >>> -- >>> Todd Lipcon >>> Software Engineer, Cloudera >>> >> >> > > > -- > Todd Lipcon > Software Engineer, Cloudera > -- *MAURICIO ARISTIZABAL* Architect - Business Intelligence + Data Science mauricio@impactradius.com(m)+1 323 309 4260 223 E. De La Guerra St. | Santa Barbara, CA 93101 Overview | Twitter | Facebook | LinkedIn --f403045ea6c4bf74c305678eaf81 Content-Type: text/html; charset="UTF-8" Content-Transfer-Encoding: quoted-printable
@Todd I know working with parquet in the past I've see= n small dimensions that fit in 1 single file/block limit parallelism of joi= n/exchange/aggregation nodes, and I've forced those dims to spread acro= ss 20 or so blocks by leveraging SET PARQUET_FILE_SIZE=3D8m; or similar whe= n doing INSERT OVERWRITE to load them, which then allows these operations t= o parallelize across that many nodes.

Wouldn't it be= useful here for Cliff's small dims to be partitioned into a couple tab= lets to similarly improve parallelism?

-m

On Fri, Mar 16= , 2018 at 2:29 PM, Todd Lipcon <todd@cloudera.com> wrote:
On Fri, Mar 16, 2018 at 2:19 P= M, Cliff Resnick <cresny@gmail.com> wrote:
Hey Todd,

Thanks for th= at explanation, as well as all the great work you're doing=C2=A0 -- it&= #39;s much appreciated! I just have one=C2=A0last follow-up question. Readi= ng about BROADCAST operations (Kudu, Spark, Flink, etc. ) it seems the smal= ler table is always copied in its entirety BEFORE the predicate is evaluate= d.

That's not quite = true. If you have a predicate on a joined column, or on one of the columns = in the joined table, it will be pushed down to the "scan" operato= r, which happens before the "exchange". In addition, there is a f= eature called "runtime filters" that can push dynamically-generat= ed filters from one side of the exchange to the other.
=C2=A0
Bu= t since the Kudu client provides a serialized scanner as part of the ScanTo= ken API, why wouldn't Impala use that instead if=C2=A0it knows that the= table is Kudu and the query has any type of predicate? Perhaps if I hash-p= artition the table I could maybe force this (because that complicates a BRO= ADCAST)? I guess this is really a question for Impala but perhaps there is = a more basic reason.

Imp= ala could definitely be smarter, just a matter of programming Kudu-specific= join strategies into the optimizer. Today, the optimizer isn't aware o= f the unique properties of Kudu scans vs other storage mechanisms.

-Todd
=
=C2=A0

-Cliff

On Fri, Mar 1= 6, 2018 at 4:10 PM, Todd Lipcon <todd@cloudera.com> wrote:
On Fri, Mar 16, 2018 at 12:30 PM, Cliffo= rd Resnick <cresnick@mediamath.com> wrote:
I thought I had read that the Kudu client can configure a scan for CLO= SEST_REPLICA and assumed this was a way to take advantage of data collocati= on.

Yea, when a c= lient uses CLOSEST_REPLICA it will read a local one if available. However, = that doesn't influence the higher level operation of the Impala (or Spa= rk) planner. The planner isn't aware of the replication policy, so it w= ill use one of the existing supported JOIN strategies. Given statistics, it= will choose to broadcast the small table, which means that it will create = a plan that looks like:


= =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2= =A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0+----------------------= ---+
=C2=A0 =C2=A0 =C2= =A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 = =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0|=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2= =A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0|
=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 = =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 +---------->build=C2=A0 =C2=A0= =C2=A0 JOIN=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 |
=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0= =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 |=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 |= =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2= =A0 =C2=A0 =C2=A0|
=C2= =A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 = =C2=A0 |=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 |=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2= =A0 =C2=A0 =C2=A0 probe=C2=A0 =C2=A0 =C2=A0 |
=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 = =C2=A0 =C2=A0+--------------+=C2=A0 +-------------------------+
=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2= =A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0|=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 = =C2=A0 |=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 |
=C2=A0 =C2=A0 =C2=A0 =C2= =A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0| Exchange=C2=A0 =C2=A0 =C2=A0|=C2=A0= =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 |
=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2= =A0 +----+ (broadcast=C2=A0 =C2=A0|=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2= =A0 =C2=A0 =C2=A0 =C2=A0 |
=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 |=C2=A0 =C2=A0 |=C2=A0 =C2= =A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 |=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 = =C2=A0 =C2=A0 =C2=A0 =C2=A0 |
=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 |=C2=A0 =C2=A0 +---------= -----+=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 |
=C2=A0 =C2=A0 =C2=A0 =C2= =A0 =C2=A0 =C2=A0 |=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 = =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2= =A0 |
=C2=A0 =C2=A0 = =C2=A0 +---------+=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 = =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 |
=C2=A0 =C2=A0 =C2=A0 |=C2=A0 =C2= =A0 =C2=A0 =C2=A0 =C2=A0|=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 = =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 +-----------------------+
=C2=A0 =C2=A0 =C2=A0 |=C2=A0 SCAN=C2= =A0 =C2=A0|=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 = =C2=A0 =C2=A0 =C2=A0 |=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2= =A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0|
=C2=A0 =C2=A0 =C2=A0 |=C2=A0 KUDU=C2=A0 =C2=A0|=C2=A0 =C2=A0 =C2= =A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 |=C2=A0 = =C2=A0SCAN (other side)=C2=A0 =C2=A0|
=C2=A0 =C2=A0 =C2=A0 |=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0|= =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2= =A0 =C2=A0 |=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 = =C2=A0 =C2=A0 =C2=A0|
= =C2=A0 =C2=A0 =C2=A0 +---------+=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 = =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 +-----------------------+<= /div>

(hopefully the ASCII art comes through= )

<= /div>
In other words, the &= quot;scan kudu" operator scans the table once, and then replicates the= results of that scan into the JOIN operator. The "scan kudu" ope= rator of course will read its local copy, but it will still go through the = exchange process.

For t= he use case you're talking about, where the join is just looking up a s= ingle row by PK in a dimension table, ideally we'd be using an altogeth= er different join strategy such as nested-loop join, with the inner "l= oop" actually being a Kudu PK lookup, but that strategy isn't impl= emented by Impala.

-Todd

=C2=A0
=C2=A0If this exists then how far out of context is my understanding = of it? Reading about HDFS cache replication, I=C2=A0do know that Impala will choose a random replica there to more even= ly distribute load. But especially compared to Kudu upsert, managing mutabl= e data using Parquet is painful. So, perhaps to sum thing up, if nearly 100= % of my metadata scan are single Primary Key lookups followed by a tiny broadcast then am I really just splitting h= airs performance-wise between Kudu and HDFS-cached parquet?

From:=C2=A0 Todd Lipcon <todd@cloudera.com> Reply-To: "user@kudu.apache.org" <<= a href=3D"mailto:user@kudu.apache.org" target=3D"_blank">user@kudu.apache.o= rg>
Date: Friday, March 16, 2018 at 2:5= 1 PM

To: "user@kudu.apache.org" <user@kudu.apache.org= >
Subject: Re: "broadcast" = tablet replication for kudu?

It's worth noting that, even if your table is replicat= ed, Impala's planner is unaware of this fact and it will give the same = plan regardless. That is to say, rather than every node scanning its local = copy, instead a single node will perform the whole scan (assuming it's a small table) and broadcast it from there w= ithin the scope of a single query. So, I don't think you'll see any= performance improvements on Impala queries by attempting something like an= extremely high replication count.

I could see bumping the replication count to 5 for these tables since = the extra storage cost is low and it will ensure higher availability of the= important central tables, but I'd be surprised if there is any measura= ble perf impact.

-Todd

On Fri, Mar 16, 2018 at 11:35 AM, Clifford Resni= ck <cresnick@me= diamath.com> wrote:
Thanks for that, glad I was wrong there! Aside from replication consid= erations, is it also recommended the number of tablet servers be odd?

I will check forums as you suggested, but from what I read after searc= hing is that Impala relies on user configured caching strategies using HDFS= cache.=C2=A0 The workload for these tables is very light write, maybe a do= zen or so records per hour across 6 or 7 tables. The size of the tables ranges from thousands to low millions of = rows so so sub-partitioning would not be required. So perhaps this is not a= typical use-case but I think it could work quite well with kudu.

From: Dan Burkert <danburkert@apache.org>=
Reply-To: "user@kudu.apache.org" <<= a href=3D"mailto:user@kudu.apache.org" target=3D"_blank">user@kudu.apache.o= rg>
Date: Friday, March 16, 2018 at 2:0= 9 PM
To: "user@kudu.apache.org" <user@kudu.apache.org= >
Subject: Re: "broadcast" = tablet replication for kudu?

The replication count is the number of tablet servers which Kudu will = host copies on.=C2=A0 So if you set the replication level to 5, Kudu will p= ut the data on 5 separate tablet servers.=C2=A0 There's no built-in bro= adcast table feature; upping the replication factor is the closest thing.=C2=A0 A couple of things to keep in mind:

- Always use an odd replication count.=C2=A0 This is important due to how t= he Raft algorithm works.=C2=A0 Recent versions of Kudu won't even let y= ou specify an even number without flipping some flags.
- We don't test much much beyond 5 replicas.=C2=A0 It should wor= k, but you may run in to issues since it's a relatively rare configurat= ion.=C2=A0 With a heavy write workload and many replicas you are even more = likely to encounter issues.

It's also worth checking in an Impala forum whether it has feature= s that make joins against small broadcast tables better?=C2=A0 Perhaps Impa= la can cache small tables locally when doing joins.

- Dan

On Fri, Mar 16, 2018 at 10:55 AM, Clifford Resni= ck <cresnick@me= diamath.com> wrote:
The problem is, AFIK, that replication count is not neces= sarily the distribution count, so you can't guarantee all tablet server= s will have a copy.

On Mar 16, 2018 1:41 PM, Boris Tyukin <boris@boristyukin.com<= /a>> wrote:
I'm new to Kudu but we are also going to use Impala mo= stly with Kudu. We have a few tables that are small but used a lot. My plan= is replicate them more than 3 times. When you create a kudu table, you can= specify number of replicated copies (3 by default) and I guess you can put there a number, corresponding to your = node count in cluster. The downside, you cannot change that number unless y= ou recreate a table.

On Fri, Mar 16, 2018 at 10:42 AM, Cliff Resnick = <cresny@gmail.com<= /a>> wrote:
We will soon be moving our analytics from AWS Redshift to = Impala/Kudu. One Redshift feature that we will miss is its ALL Distribution= , where a copy of a table is maintained on each server. We define a number = of metadata tables this way since they are used in nearly every query. We are considering using parquet in H= DFS cache for these, and Kudu would be a much better fit for the update sem= antics but we are worried about the additional contention.=C2=A0 I'm wo= ndering if having a Broadcast, or ALL, tablet replication might be an easy feature to add to Kudu?=C2=A0

-Cliff=C2=A0





--
Todd Li= pcon
Software Engineer, Cloudera



--
Todd Lipcon
Software Engineer, Cloudera<= /div>




--
Todd Lipcon
Software Engi= neer, Cloudera



--
--f403045ea6c4bf74c305678eaf81--