在apache pig中将键值元组包转换为Map

drnojrws 于 2021-06-24 发布在 Pig

关注(0)|答案(2)|浏览(438)

我对pig很陌生，我想把一包元组转换成一个Map，每个元组中都有特定的值作为键。基本上我想改变： {(id1, value1),(id2, value2), ...} 进入 [id1#value1, id2#value2] 我在网上找了一段时间，但似乎找不到解决办法。我试过：

bigQMap = FOREACH bigQFields GENERATE TOMAP(queryId, queryStart);

但我最终得到了一袋Map（例如。 {[id1#value1], [id2#value2], ...} )，这不是我想要的。如何从一袋键值元组中建立一个Map？
下面是我试图运行的特定脚本，以防相关

rawlines = LOAD '...' USING PigStorage('`');
bigQFields = FOREACH bigQLogs GENERATE GFV(*,'queryId')
   as queryId, GFV(*, 'queryStart')
   as queryStart;
bigQMap = ?? how to make a map with queryId as key and queryStart as value ?? ;

apache-pig Map

来源：https://stackoverflow.com/questions/17847970/transform-bag-of-key-value-tuples-to-map-in-apache-pig

2条答案

按热度按时间

u2nhd7ah1#

TOMAP 获取一系列对并将它们转换为Map，因此它的用法如下：

-- Schema: A:{foo:chararray, bar:int, bing:chararray, bang:int}
-- Data:     (John,          27,      Joe,            30)
B = FOREACH A GENERATE TOMAP(foo, bar, bing, bang) AS m ;
-- Schema: B:{m: map[]}
-- Data:     (John#27,Joe#30)

因此，正如您所见，语法不支持将包转换为Map。据我所知，没有办法将一个包转换成你必须在纯PigMap的格式。但是，您可以明确地编写一个javaudf来实现这一点。
注意：我对java不是很有经验，所以这个udf可以很容易地改进（添加异常处理，如果一个键添加两次会发生什么等等）。然而，它确实完成了你所需要的。

package myudfs;
import java.io.IOException;
import org.apache.pig.EvalFunc;
import java.util.Map;
import java.util.HashMap;
import java.util.Iterator;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.DataBag;
public class ConvertToMap extends EvalFunc<Map>
{
    public Map exec(Tuple input) throws IOException {
        DataBag values = (DataBag)input.get(0);
        Map<Object, Object> m = new HashMap<Object, Object>();
        for (Iterator<Tuple> it = values.iterator(); it.hasNext();) {
            Tuple t = it.next();
            m.put(t.get(0), t.get(1));
        }
        return m;
    }
}

一旦将脚本编译到jar中，就可以像这样使用它：

REGISTER myudfs.jar ;
-- A is loading some sample data I made
A = LOAD 'foo.in' AS (foo:{T:(id:chararray, value:chararray)}) ;
B = FOREACH A GENERATE myudfs.ConvertToMap(foo) AS bar;

的内容 foo.in :

{(open,apache),(apache,hadoop)}
{(foo,bar),(bar,foo),(open,what)}

输出自 B :

([open#apache,apache#hadoop])
([bar#foo,open#what,foo#bar])

另一种方法是使用python创建udf：

我的自定义项.py


# !/usr/bin/python
@outputSchema("foo:map[]")
def BagtoMap(bag):
    d = {}
    for key, value in bag:
        d[key] = value
    return d

用法如下：

Register 'myudfs.py' using jython as myfuncs;
-- A is still just loading some of my test data
A = LOAD 'foo.in' AS (foo:{T:(key:chararray, value:chararray)}) ;
B = FOREACH A GENERATE myfuncs.BagtoMap(foo) ;

并生成与javaudf相同的输出。
优点：因为我不太喜欢Map，这里有一个链接解释了如何用键值对复制Map的功能。因为您的键值对在一个包中，所以您需要在嵌套的 FOREACH :

-- A is a schema that contains kv_pairs, a bag in the form {(id, value)}
B = FOREACH A {
    temp = FOREACH kv_pairs GENERATE (key=='foo'?value:NULL) ;
    -- Output is like: ({(),(thevalue),(),()})
    -- MAX will pull the maximum value from the filtered bag, which is 
    -- value (the chararray) if the key matched. Otherwise it will return NULL.
    GENERATE MAX(temp) as kv_pairs_filtered ;
}