6

我是 pig 新手，在解析我的输入并将其转换为我可以使用的格式时遇到问题。输入文件包含具有固定字段和 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);

这会生成具有如下模式的记录：

A：{日期：long，hms：long，id：long，ff1：chararray，ff2：long，ff3：chararray，ff4：chararray，kvpairs：{kvpair：（NULL）}}

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

  1. 通过将 TOBAG 与 .. 一起使用，没有架构可以应用于我的 kvpair，因此我无法过滤 kvpair，而且我似乎无法在任何时候强制转换，所以这是一个全有或全无的查询.
  2. 无论我使用什么值，语句“C”中的过滤器似乎都不会返回任何数据，即使我使用了“.*”或“.+”之类的东西。我不知道这是因为没有架构，还是这实际上是 pig 中的错误。如果我从语句 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

提前致谢。

4

1 回答 1

2

你的第二种方法是在正确的轨道上。不幸的是，你需要一个 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变成一个袋子，把它压平，然后进行分组和计数。

于 2013-09-27T19:13:50.223 回答