hadoop 猪包减法答案

【问题标题】：hadoop pig bag subtractionhadoop 猪包减法
【发布时间】：2012-08-16 15:23:29
【问题描述】：

我正在使用 Pig 解析我的应用程序日志，以了解上个月未被调用的用户（由同一用户）调用了哪些公开的方法。

我已经设法在上个月之前和上个月之后按用户分组调用方法：

上个月之前的关系样本

u1      {(m1),(m2)}
u2      {(m3),(m4)}

上个月之后的关系样本

u1      {(m1),(m3)}
u2      {(m1),(m4)}

我想要的是由用户找到哪些方法在AFTER中而不在BEFORE中，即

NEWLY_CALLED 预期结果

u1      {(m3)}
u2      {(m1)}

问题：我如何在 Pig 中做到这一点？可以减袋吗？

我尝试了 DIFF 函数，但它没有执行预期的减法。

问候，

乔尔

【问题讨论】：

标签： hadoop apache-pig

【解决方案1】：

我认为你需要写一个UDF，然后你可以使用

Set<T> setA ...
Set<T> setB ...
Set<T> setAminusB = setA.subtract(setB);

【讨论】：

几分钟前我刚刚这样做了 :) 感谢 Mark 的建议！我将提出我对猪/存钱罐所做的事情，因为我认为这可能对其他人有所帮助。
@JoelCostigliola 有内置函数SUBTRACT。这是你需要的吗？
@wenlong 在 pig 0.11.1 中是否支持 SUBTRACT？

【解决方案2】：

对于那些可能感兴趣的人，这是我在下面编写的类并提出给 Pig (PIG-2881) 的减法函数：

/**
 * Subtract takes two bags as arguments returns a new bag composed of tuples of first bag not in the second bag.<br>
 * If null bag arguments are replaced by empty bags. 
 * <p>
 * The implementation assumes that both bags being passed to this function will fit entirely into memory simultaneously.
 * </br>
 * If that is not the case the UDF will still function, but it will be <strong>very</strong> slow.
 */
public class Subtract extends EvalFunc<DataBag> {

  /**
   * Compares the two bag fields from input Tuple and returns a new bag composed of elements of first bag not in the second bag.
   * @param input a tuple with exactly two bag fields.
   * @throws IOException if there are not exactly two fields in a tuple or if they are not {@link DataBag}.
   */
  @Override
  public DataBag exec(Tuple input) throws IOException {
    if (input.size() != 2) {
      throw new ExecException("Subtract expected two inputs but received " + input.size() + " inputs.");
    }
    DataBag bag1 = toDataBag(input.get(0));
    DataBag bag2 = toDataBag(input.get(1));
    return subtract(bag1, bag2);
  }

  private static String classNameOf(Object o) {
    return o == null ? "null" : o.getClass().getSimpleName();
  }

  private static DataBag toDataBag(Object o) throws ExecException {
    if (o == null) {
      return BagFactory.getInstance().newDefaultBag();
    }
    if (o instanceof DataBag) {
      return (DataBag) o;
    }
    throw new ExecException(format("Expecting input to be DataBag only but was '%s'", classNameOf(o)));
  }

  private static DataBag subtract(DataBag bag1, DataBag bag2) {
    DataBag subtractBag2FromBag1 = BagFactory.getInstance().newDefaultBag();
    // convert each bag to Set,  this does make the assumption that the sets will fit in memory.
    Set<Tuple> set1 = toSet(bag1);
    // remove elements of bag2 from set1 
    Iterator<Tuple> bag2Iterator = bag2.iterator();
    while (bag2Iterator.hasNext()) {
      set1.remove(bag2Iterator.next());
    }
    // set1 now contains all elements of bag1 not in bag2 => we can build the resulting DataBag.
    for (Tuple tuple : set1) {
      subtractBag2FromBag1.add(tuple);
    }
    return subtractBag2FromBag1;
  }

  private static Set<Tuple> toSet(DataBag bag) {
    Set<Tuple> set = new HashSet<Tuple>();
    Iterator<Tuple> iterator = bag.iterator();
    while (iterator.hasNext()) {
      set.add(iterator.next());
    }
    return set;
  }

}

【讨论】：