hadoop - Hadoop 中的自定义分区程序

Question

我有一些数据由 0 到 200 之间的 id 键控 - 大约一百万，我需要将它分成 0-500 万、500 万 - 1000 万等范围的美元。

我正在尝试在 Hadoop 上为最后一部分使用自定义分区程序，以便我的代码的最后一部分看起来像这样:

Conns = FOREACH ConnsGrouped GENERATE group as memberId, $1.companyId as companyIds;
ConnsPartitioned = DISTINCT Conns PARTITION BY com.mypackage.SearchNodePartitioner PARALLEL 50;

rmf $connections_file

Store ConnsPartitioned INTO 'test' using AvroStorage(...);

我的分区器看起来像这样:

public class SearchNodePartitioner<Long, V> implements Partitioner<Long, V>
{
    @Override
    public void configure(JobConf conf) 
    {
        // Nothing
    }

    @Override
    public int getPartition(Long key, V value, int numPartitions) 
    {
       return new Double(Math.floor(key / (5.0 * Math.pow(10, 6)))).intValue() % numPartitions;
    }

}

但是好像根本就没有被调用。即使我用 return 1; 替换 return 行，跨文件的数据似乎是散列分布的默认行为。

Answer 1

DISTINCT + 自定义分区程序的答案是:你不能再这样做了(正如我刚刚发现的那样)。 DISTINCT 现在使用优化的特殊分区程序。

参见:

http://mail-archives.apache.org/mod_mbox/pig-user/201307.mbox/%3C14FE3AC3-DBA5-4898-AF94-0C34819A0D8B%40hortonworks.com%3E

https://issues.apache.org/jira/browse/PIG-3385

解决方法:

A =//一些元组...;

B = GROUP A BY 字段 PARTITION BY 自定义；

使用 .... 将 B 存储到“foo”中；

后来:

B = 使用 ... 加载“foo”；

A = FOREACH B 生成扁平化($1)；