Return-Path: X-Original-To: apmail-hadoop-hdfs-user-archive@minotaur.apache.org Delivered-To: apmail-hadoop-hdfs-user-archive@minotaur.apache.org Received: from mail.apache.org (hermes.apache.org [140.211.11.3]) by minotaur.apache.org (Postfix) with SMTP id 0673695AD for ; Thu, 25 Oct 2012 07:17:54 +0000 (UTC) Received: (qmail 5845 invoked by uid 500); 25 Oct 2012 07:17:48 -0000 Delivered-To: apmail-hadoop-hdfs-user-archive@hadoop.apache.org Received: (qmail 5644 invoked by uid 500); 25 Oct 2012 07:17:48 -0000 Mailing-List: contact user-help@hadoop.apache.org; run by ezmlm Precedence: bulk List-Help: List-Unsubscribe: List-Post: List-Id: Reply-To: user@hadoop.apache.org Delivered-To: mailing list user@hadoop.apache.org Received: (qmail 5633 invoked by uid 99); 25 Oct 2012 07:17:48 -0000 Received: from athena.apache.org (HELO athena.apache.org) (140.211.11.136) by apache.org (qpsmtpd/0.29) with ESMTP; Thu, 25 Oct 2012 07:17:48 +0000 X-ASF-Spam-Status: No, hits=1.5 required=5.0 tests=HTML_MESSAGE,RCVD_IN_DNSWL_LOW,SPF_PASS X-Spam-Check-By: apache.org Received-SPF: pass (athena.apache.org: domain of dechouxb@gmail.com designates 209.85.216.176 as permitted sender) Received: from [209.85.216.176] (HELO mail-qc0-f176.google.com) (209.85.216.176) by apache.org (qpsmtpd/0.29) with ESMTP; Thu, 25 Oct 2012 07:17:43 +0000 Received: by mail-qc0-f176.google.com with SMTP id n41so595243qco.35 for ; Thu, 25 Oct 2012 00:17:22 -0700 (PDT) DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=gmail.com; s=20120113; h=mime-version:in-reply-to:references:date:message-id:subject:from:to :content-type; bh=+vWbDON5JvUQaCJoHnsBMqp39Lq27ew/7DrAq6mjAyU=; b=pGssUgUnxIaqSWp0cotoFY7T1Xd4P3d6oB3PDUah98kxnQwAVY79+/0Eb6mtB07+xV LgHp+F0DZ3N+o0EUn4LcWAm5QEc7gLNuk84wmCg6RFozi1K8O7WgoC6CgPNSMxxiUgx7 S7WjZVw7W7vCrvYkBVNaCw4GyIuGV8JAwsSg0IoQrqGZjvltyeXLzKkMFePCOLjgvWTF 7tgbmXsfg6oW/It9FM5fEKvpzk6G5CVk0RlGWzZP3R99tRmVOcM3SFvOym+OTlSsEjwj VcbNJXc4h2S+Ld9bdqRwGEjRWL81WsgcZaeK+P4Ovu9j2vaHvinS/sgbZA32Bkf6U5Rl 0VFQ== MIME-Version: 1.0 Received: by 10.224.117.20 with SMTP id o20mr5766276qaq.12.1351149442356; Thu, 25 Oct 2012 00:17:22 -0700 (PDT) Received: by 10.49.71.231 with HTTP; Thu, 25 Oct 2012 00:17:22 -0700 (PDT) In-Reply-To: References: <5085AC9B.4070504@gmail.com> <1230403896-1350973318-cardhu_decombobulator_blackberry.rim.net-1541577741-@b27.c16.bise7.blackberry> Date: Thu, 25 Oct 2012 09:17:22 +0200 Message-ID: Subject: Re: Data locality of map-side join From: Bertrand Dechoux To: user@hadoop.apache.org Content-Type: multipart/alternative; boundary=20cf3074b3aa8cd66a04ccdcfe46 X-Virus-Checked: Checked by ClamAV on apache.org --20cf3074b3aa8cd66a04ccdcfe46 Content-Type: text/plain; charset=ISO-8859-1 One underlying issue is that you would like your tool to be able to detect which dataset is the largest and how large is it because with this information different strategies can be chosen. This implies that somehow your tool needs to create/keep/update statistics about your datasets. And that's clearly something which is relevant for an external tool (like hive or pig) but it might not make sense to build that into the core mapred/mapreduce. That would increase coupling for something which is not necessarily relevant for the core of the platform. I know about about Hive. And you could be interested in reading more about it. https://cwiki.apache.org/Hive/statsdev.html Statistics such as the number of rows of a table or partition and the > histograms of a particular interesting column are important in many ways. > One of the key use cases of statistics is query optimization. Statistics > serve as the input to the cost functions of the optimizer so that it can > compare different plans and choose among them. Statistics may sometimes > meet the purpose of the users' queries. Users can quickly get the answers > for some of their queries by only querying stored statistics rather than > firing long-running execution plans. Some examples are getting the quantile > of the users' age distribution, the top 10 apps that are used by people, > and the number of distinct sessions. > I don't know if Pig has something similar. Regards Bertrand On Thu, Oct 25, 2012 at 7:49 AM, Harsh J wrote: > Hi Sigurd, > > From what I've generally noticed, the client-end frameworks (Hive, > Pig, etc.) have gotten much more cleverness and efficiency packed in > their join parts than the MR join package which probably exists to > serve as an example or utility today more than anything else (but > works well for what it does). > > Per the code in the join package, there are no such estimates made > today. There is zero use of DistributedCache - the only decisions are > made based on the expression (i.e. to select which form of joining > record reader to use). > > Enhancements to this may be accepted though, so feel free to file some > JIRAs if you have something to suggest/contribute. Hopefully one day > we could have a unified library between client-end tools for common > use-cases such as joins, etc. over MR, but there isn't such a thing > right now (AFAIK). > > On Tue, Oct 23, 2012 at 2:52 PM, Sigurd Spieckermann > wrote: > > Interesting to know that Hive and Pig are doing something in this > direction. > > I'm dealing with the Hadoop join-package which doesn't use > DistributedCache > > though but it rather pulls the other partition over the network before > > launching the map task. This is under the assumption that both partitions > > are too big to load into DC or it's just undesirable to use DC. Is there > a > > similar mechanism implemented in the join-package that considers the > size of > > the two partitions to be joined trying to execute the map task on the > > datanode that holds the bigger partition? > > > > > > 2012/10/23 Bejoy KS > >> > >> Hi Sigurd > >> > >> Mapside joins are efficiently implemented in Hive and Pig. I'm talking > in > >> terms of how mapside joins are implemented in hive. > >> > >> In map side join, the smaller data set is first loaded into > >> DistributedCache. The larger dataset is streamed as usual and the > smaller > >> dataset in memory. For every record in larger data set the look up is > made > >> in memory on the smaller set and there by joins are done. > >> > >> In later versions of hive the hive framework itself intelligently > >> determines the smaller data set. In older versions you can specify the > >> smaller data set using some hints in query. > >> > >> > >> Regards > >> Bejoy KS > >> > >> Sent from handheld, please excuse typos. > >> > >> -----Original Message----- > >> From: Sigurd Spieckermann > >> Date: Mon, 22 Oct 2012 22:29:15 > >> To: > >> Reply-To: user@hadoop.apache.org > >> Subject: Data locality of map-side join > >> > >> Hi guys, > >> > >> I've been trying to figure out whether a map-side join using the > >> join-package does anything clever regarding data locality with respect > >> to at least one of the partitions to join. To be more specific, if I > >> want to join two datasets and some partition of dataset A is larger than > >> the corresponding partition of dataset B, does Hadoop account for this > >> and try to ensure that the map task is executed on the datanode storing > >> the bigger partition thus reducing data transfer (if the other partition > >> does not happen to be located on that same datanode)? I couldn't > >> conclude the one or the other behavior from the source code and I > >> couldn't find any documentation about this detail. > >> > >> Thanks for clarifying! > >> Sigurd > > > > > > > > -- > Harsh J > -- Bertrand Dechoux --20cf3074b3aa8cd66a04ccdcfe46 Content-Type: text/html; charset=ISO-8859-1 Content-Transfer-Encoding: quoted-printable One underlying issue is that you would like your tool to be able to detect = which dataset is the largest and how large is it because with this informat= ion different strategies can be chosen. This implies that somehow your tool= needs to create/keep/update statistics about your datasets. And that's= clearly something which is relevant for an external tool (like hive or pig= ) but it might not make sense to build that into the core mapred/mapreduce.= That would increase coupling for something which is not necessarily releva= nt for the core of the platform.

I know about about Hive. And you could be interested in reading more ab= out it.
https://= cwiki.apache.org/Hive/statsdev.html

Statistics such as the number of rows of a table or partition and the=20 histograms of a particular interesting column are important in many=20 ways. One of the key use cases of statistics is query optimization.=20 Statistics serve as the input to the cost functions of the optimizer so=20 that it can compare different plans and choose among them. Statistics=20 may sometimes meet the purpose of the users' queries. Users can quickly= =20 get the answers for some of their queries by only querying stored=20 statistics rather than firing long-running execution plans. Some=20 examples are getting the quantile of the users' age distribution, the= =20 top 10 apps that are used by people, and the number of distinct=20 sessions.

I don't know if Pig has something si= milar.

Regards

Bertrand


On Thu, Oct 25, 2012 at 7:49 AM, Harsh J <harsh@cloudera.com&= gt; wrote:
Hi Sigurd,

>From what I've generally noticed, the client-end frameworks (Hive,
Pig, etc.) have gotten much more cleverness and efficiency packed in
their join parts than the MR join package which probably exists to
serve as an example or utility today more than anything else (but
works well for what it does).

Per the code in the join package, there are no such estimates made
today. There is zero use of DistributedCache - the only decisions are
made based on the expression (i.e. to select which form of joining
record reader to use).

Enhancements to this may be accepted though, so feel free to file some
JIRAs if you have something to suggest/contribute. Hopefully one day
we could have a unified library between client-end tools for common
use-cases such as joins, etc. over MR, but there isn't such a thing
right now (AFAIK).

On Tue, Oct 23, 2012 at 2:52 PM, Sigurd Spieckermann
<sigurd.spieckermann@gm= ail.com> wrote:
> Interesting to know that Hive and Pig are doing something in this dire= ction.
> I'm dealing with the Hadoop join-package which doesn't use Dis= tributedCache
> though but it rather pulls the other partition over the network before=
> launching the map task. This is under the assumption that both partiti= ons
> are too big to load into DC or it's just undesirable to use DC. Is= there a
> similar mechanism implemented in the join-package that considers the s= ize of
> the two partitions to be joined trying to execute the map task on the<= br> > datanode that holds the bigger partition?
>
>
> 2012/10/23 Bejoy KS <bejo= y.hadoop@gmail.com>
>>
>> Hi Sigurd
>>
>> Mapside joins are efficiently implemented in Hive and Pig. I'm= talking in
>> terms of how mapside joins are implemented in hive.
>>
>> In map side join, the smaller data set is first loaded into
>> DistributedCache. The larger dataset is streamed as usual and the = smaller
>> dataset in memory. For every record in larger data set the look up= is made
>> in memory on the smaller set and there by joins are done.
>>
>> In later versions of hive the hive framework itself intelligently<= br> >> determines the smaller data set. In older versions you can specify= the
>> smaller data set using some hints in query.
>>
>>
>> Regards
>> Bejoy KS
>>
>> Sent from handheld, please excuse typos.
>>
>> -----Original Message-----
>> From: Sigurd Spieckermann <sigurd.spieckermann@gmail.com>
>> Date: Mon, 22 Oct 2012 22:29:15
>> To: <user@hadoop.apac= he.org>
>> Reply-To: user@hadoop.ap= ache.org
>> Subject: Data locality of map-side join
>>
>> Hi guys,
>>
>> I've been trying to figure out whether a map-side join using t= he
>> join-package does anything clever regarding data locality with res= pect
>> to at least one of the partitions to join. To be more specific, if= I
>> want to join two datasets and some partition of dataset A is large= r than
>> the corresponding partition of dataset B, does Hadoop account for = this
>> and try to ensure that the map task is executed on the datanode st= oring
>> the bigger partition thus reducing data transfer (if the other par= tition
>> does not happen to be located on that same datanode)? I couldn'= ;t
>> conclude the one or the other behavior from the source code and I<= br> >> couldn't find any documentation about this detail.
>>
>> Thanks for clarifying!
>> Sigurd
>
>



--
Harsh J



--
Bertrand = Dechoux
--20cf3074b3aa8cd66a04ccdcfe46--