我已经尝试了一段时间，似乎无法解决这个问题，这是有点难以找到一个解决我的问题。
我有一个关系，我以前分组 user_id 以及 listing_id 在生成并展平输出之后，我得到了：

test: {user_id: bytearray,listing_id: bytearray,hotness: long}

因此，我的下一步是按用户分组，按热度排序，并将每个用户的列表数量限制为20个。

grped = GROUP test BY user_id;
grped_sorted = FOREACH grped {
    sorted = order test BY hotness desc;
    top1   = limit sorted 20;
    listings = FOREACH top1 GENERATE FLATTEN((bytearray)top1.listing_id) as listing_id;
         GENERATE group as user_id, FLATTEN(listings.($0)) as listing_ids;
   };

但这似乎给我带来了一个错误，之前从清单细节中删除的信息：

Scalar has more than one row in the output.

拜托，我需要帮助。有办法吗？我可以用datafu的自定义项吗？创建自己的自定义项是不可能的。
提前谢谢。