使用pig,如何将混合格式行解析为元组和一包元组？

Question

我是猪的新手,我在解析输入并将其转换为我可以使用的格式时遇到问题.输入文件包含具有固定字段和KV对的行,如下所示:

FF1 | FF2 | FF3 | FF4 | KVP1 | KVP2 | ... | KVPn

我的目标是计算每个KV对的唯一固定字段组合的数量.所以考虑以下输入行:

1|2|3|4|key1=value1|key2=value2
2|3|4|5|key1=value7|key2=value2|key3=value3

当我完成后,我希望能够生成以下结果(此时输出格式并不重要,我只是向您展示我想要的结果):

key1=value1 : 1
key1=value7 : 1
key2=value2 : 2
key3=value3 : 1

似乎我应该能够通过将固定区域分组并展平一袋KV对来生成交叉产品

我试过用以下的东西读这个:

data = load 'myfile' using PigStorage('|');
A = foreach data generate $0 as ff1:chararray, $1 as ff2:long, $2 as ff3:chararray, $3 as ff4:chararray, TOBAG($4..) as kvpairs:bag{kvpair:tuple()};
B = foreach A { sorted = order A by ff2; lim = limit sorted 1; generate group.ff1, group.ff4, flatten( lim.kvpairs ); };
C = filter B by ff3 matches 'somevalue';
D = foreach C generate ff1, ff4, flatten( kvpairs ) as kvpair;
E = group D by (ff1, ff4, kvpair);
F = foreach E generate group, COUNT(E);

这将生成具有模式的记录,如下所示:

答:{date:long,hms:long,id:long,ff1:chararray,ff2:long,ff3:chararray,ff4:chararray,kvpairs:{kvpair:(NULL)}}

虽然这让我得到了我想要的架构,但有几个问题似乎无法解决:

1. 通过使用TOBAG和..,没有架构可以应用到我的kvpairs,所以我不能过滤kvpair,我似乎无法在任何时候转换它,所以这是一个全有或全无的查询.
2. 语句'C'中的过滤器似乎不返回任何数据,无论我使用什么值,即使我使用'.*'或'.+'之类的东西.我不知道这是因为没有架构,或者这实际上是猪的错误.如果我从语句B转储一些数据,我肯定会看到那些与那些表达式匹配的数据.

所以我尝试通过以下方式加载数据来以不同方式解决问题:

data = load 'myfile' using PigStorage('\n') as (line:chararray);
init_parse = foreach data generate FLATTEN( STRSPLIT( line, '\\|', 4 ) ) as (ff1:chararray, ff2:chararray, ff3:chararray, ff4:chararray, kvpairsStr:chararray);
A = foreach mc_bk_data generate ff1, ff2, ff3, ff4, TOBAG( STRSPLIT( kvpairsStr, '\\|', 500 ) ) as kvpairs:bag{t:(kvpair:chararray)};

这里的问题是TOBAG(STRSPLIT(...))产生一个单元组的包,每个kvpairs都是该元组中的一个字段.我真的需要把包包起来,每个单独的kvpairs作为一个字段的元组,这样当我压扁包时,我得到了包和我感兴趣的组的交叉产品.

我也对其他攻击这个问题的方式持开放态度,但我似乎找到了将多个字段的元组转换成一个元组包的好方法,每个元组都有一个字段.

我正在使用Apache Pig版本0.11.1.1.3.0.0-107

提前致谢.

Answer 1

你的第二种方法是在正确的轨道上。不幸的是，您需要一个 UDF 将元组转换为包，据我所知，没有内置函数可以执行此操作。然而，写一个是一件简单的事情。

您不想对固定字段进行分组，而是对键值对本身进行分组。所以只需要保留键值对的元组即可；您可以完全忽略固定字段。

UDF 非常简单。在Java中，你可以在你的exec方法中做这样的事情：

DataBag b = new DefaultDataBag();
Tuple t = (Tuple) input.get(0);
for (int i = 0; i < t.size(); i++) {
    Object o = t.get(i);
    Tuple e = TupleFactory.getInstance().createTuple(o);
    b.add(e);
}

return b;

获得后，将元组放入STRSPLIT袋子中，将其压平，然后进行分组和计数。