hadoop - Pig 0.11.0 在应用 DISTINCT 之前不排序

Question

我有一些包含一些包作为字段的记录，我正在尝试合并具有其他相同字段的记录的包(我正在丢弃一些字段)。

数据看起来像这样:

u08 u08an   id  {(web)} 0   0   {(GB),(US)} an
u08 u08an   id  {(ars)} 0   0   {(GB),(RU)} an
u09 u09an   id  {(web)} 0   0   {(GB)}  an
u09 u09an   id  {(web)} 0   0   {(US)}  an
u10 u10an   id  {(web)} 0   0   {(GB)}  an
u10 u10an   id  {(ars)} 0   0   {(GB)}  an
u11 u11an   id  {(web)} 0   0   {(GB)}  an
u11 u11an   id  {(web)} 0   0   {(GB)}  an

我想获得(在丢弃不相关的字段和洗牌之后)类似的东西:

u08 u08an   an  {(GB),(US),(RU)}
u09 u09an   an  {(GB),(US)}
u10 u10an   an  {(GB)}
u11 u11an   an  {(GB)}

使用以下模式加载输入:

user_identities = LOAD '$partner_user_identities_location' AS (
    user_id: chararray,
    partner_user_id: chararray,
    partner_user_id_type: chararray,
    sync_types: bag{tuple(chararray)},
    synced_first_timestamp: double,
    synced_last_timestamp: double,
    country_codes: bag{tuple(chararray)},
    partner_id: chararray
);

如果我合并 sync_types 和 country_codes，一切都会按预期工作，但如果我只生成 country_codes，则记录不会在应用 DISTINCT 之前排序，因此不相邻的重复项保留在输出中。

运行以下代码片段(在本地模式下):

user_identities = GROUP user_identities BY (user_id, partner_user_id, partner_id);

user_identities = FOREACH user_identities {
    sync_types = FOREACH user_identities GENERATE flatten(sync_types);
    sync_types = DISTINCT sync_types;

    country_codes = FOREACH user_identities GENERATE flatten(country_codes);
    country_codes = DISTINCT country_codes;

    GENERATE flatten(group) AS (user_id, partner_user_id, partner_id), sync_types, country_codes;
}

DUMP user_identities;

输出:

(u08,u08an,an,{(ars),(web)},{(GB),(RU),(US)})
(u09,u09an,an,{(web)},{(GB),(US)})
(u10,u10an,an,{(ars),(web)},{(GB)})
(u11,u11an,an,{(web)},{(GB)})

但是，如果我将内部 GENERATE 语句更改为 GENERATE flatten(group) AS (user_id, partner_user_id, partner_id), country_codes;，省略 sync_types，我得到以下输出(注意第一条记录中重复的 (GB)):

(u08,u08an,an,{(GB),(RU),(GB),(US)})
(u09,u09an,an,{(GB),(US)})
(u10,u10an,an,{(GB)})
(u11,u11an,an,{(GB)})

因为我没有在我的脚本中使用 sync_types，所以除了作为解决此问题的解决方法外，我看不出为什么要生成它。

这是 Pig(或 Pig 的本地模式)中的已知错误吗？还是我没有正确合并袋子？

Answer 1

我不确定您是否仍在寻找答案。 我在 Pig 版本 0.12.x 中使用了 DISTINCT，它似乎按预期工作，但是我修改了如下查询以实现您的预​​期结果 -

user_identities = LOAD 'userPig.txt' AS (
    user_id: chararray,
    partner_user_id: chararray,
    partner_user_id_type: chararray,
    sync_types: bag{tuple(chararray)}, 
    synced_first_timestamp: double,
    synced_last_timestamp: double,
    country_codes: bag{tuple(chararray)},
    partner_id: chararray);

    user_identities_filter = foreach user_identities generate user_id,
        partner_user_id,  partner_id, country_codes;

    user_identities_group = GROUP user_identities_filter BY (user_id,
        partner_user_id, partner_id);

    user_stats = FOREACH user_identities_group {
       uniq_country_codes = FOREACH user_identities_filter GENERATE  
           flatten(country_codes);
       uniq_country_codes = DISTINCT uniq_country_codes;
       GENERATE FLATTEN(group), uniq_country_codes AS uniq_coutry_codes;
    }

    DUMP user_stats;