面试官问：UDF是在Map端执行还是Reduce端执行?

select explode(array(100,200,300));

得到的结果如下：

select posexplode(array('A','B','C'));

select parse_url_tuple('http://facebook.com/path1/p.php?k1=v1&k2=v2#Ref1','HOST', 'PATH', 'QUERY', 'QUERY:id');

 select inline(array(struct('A',10,date '2015-01-01'),struct('B',20,date '2016-02-02')));

select stack(2,'A',10,date '2015-01-01','B',20,date '2016-01-01');

 select json_tuple('{"a":1,"b":2}','a','b','c');

public class UDFYear extends UDF {
  private final SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd");
  private final Calendar calendar = Calendar.getInstance();

  private final IntWritable result = new IntWritable();

  public UDFYear() {
  }

  public IntWritable evaluate(Text dateString) {

    if (dateString == null) {
      return null;
    }

    try {
      Date date = formatter.parse(dateString.toString());
      calendar.setTime(date);
      result.set(calendar.get(Calendar.YEAR));
      return result;
    } catch (ParseException e) {
      return null;
    }
  }

  public IntWritable evaluate(DateWritable d) {
    if (d == null) {
      return null;
    }

    calendar.setTime(d.get());
    result.set(calendar.get(Calendar.YEAR));
    return result;
  }

  public IntWritable evaluate(TimestampWritable t) {
    if (t == null) {
      return null;
    }

    calendar.setTime(t.getTimestamp());
    result.set(calendar.get(Calendar.YEAR));
    return result;
  }

}

public class UDAFExampleMax extends UDAF {

  static public class MaxShortevaluator implements UDAFevaluator {
    private short mMax;
    private boolean mEmpty;

    public MaxShortevaluator() {
      super();
      init();
    }

    public void init() {
      mMax = 0;
      mEmpty = true;
    }

    public boolean iterate(ShortWritable o) {
      if (o != null) {
        if (mEmpty) {
          mMax = o.get();
          mEmpty = false;
        } else {
          mMax = (short) Math.max(mMax, o.get());
        }
      }
      return true;
    }

    public ShortWritable terminatePartial() {
      return mEmpty ? null : new ShortWritable(mMax);
    }

    public boolean merge(ShortWritable o) {
      return iterate(o);
    }

    public ShortWritable terminate() {
      return mEmpty ? null : new ShortWritable(mMax);
    }
  }

  static public class MaxIntevaluator implements UDAFevaluator {
    private int mMax;
    private boolean mEmpty;

    public MaxIntevaluator() {
      super();
      init();
    }

    public void init() {
      mMax = 0;
      mEmpty = true;
    }

    public boolean iterate(IntWritable o) {
      if (o != null) {
        if (mEmpty) {
          mMax = o.get();
          mEmpty = false;
        } else {
          mMax = Math.max(mMax, o.get());
        }
      }
      return true;
    }

    public IntWritable terminatePartial() {
      return mEmpty ? null : new IntWritable(mMax);
    }

    public boolean merge(IntWritable o) {
      return iterate(o);
    }

    public IntWritable terminate() {
      return mEmpty ? null : new IntWritable(mMax);
    }
  }

  static public class MaxLongevaluator implements UDAFevaluator {
    private long mMax;
    private boolean mEmpty;

    public MaxLongevaluator() {
      super();
      init();
    }

    public void init() {
      mMax = 0;
      mEmpty = true;
    }

    public boolean iterate(LongWritable o) {
      if (o != null) {
        if (mEmpty) {
          mMax = o.get();
          mEmpty = false;
        } else {
          mMax = Math.max(mMax, o.get());
        }
      }
      return true;
    }

    public LongWritable terminatePartial() {
      return mEmpty ? null : new LongWritable(mMax);
    }

    public boolean merge(LongWritable o) {
      return iterate(o);
    }

    public LongWritable terminate() {
      return mEmpty ? null : new LongWritable(mMax);
    }
  }

  static public class MaxFloatevaluator implements UDAFevaluator {
    private float mMax;
    private boolean mEmpty;

    public MaxFloatevaluator() {
      super();
      init();
    }

    public void init() {
      mMax = 0;
      mEmpty = true;
    }

    public boolean iterate(FloatWritable o) {
      if (o != null) {
        if (mEmpty) {
          mMax = o.get();
          mEmpty = false;
        } else {
          mMax = Math.max(mMax, o.get());
        }
      }
      return true;
    }

    public FloatWritable terminatePartial() {
      return mEmpty ? null : new FloatWritable(mMax);
    }

    public boolean merge(FloatWritable o) {
      return iterate(o);
    }

    public FloatWritable terminate() {
      return mEmpty ? null : new FloatWritable(mMax);
    }
  }

  static public class MaxDoubleevaluator implements UDAFevaluator {
    private double mMax;
    private boolean mEmpty;

    public MaxDoubleevaluator() {
      super();
      init();
    }

    public void init() {
      mMax = 0;
      mEmpty = true;
    }

    public boolean iterate(DoubleWritable o) {
      if (o != null) {
        if (mEmpty) {
          mMax = o.get();
          mEmpty = false;
        } else {
          mMax = Math.max(mMax, o.get());
        }
      }
      return true;
    }

    public DoubleWritable terminatePartial() {
      return mEmpty ? null : new DoubleWritable(mMax);
    }

    public boolean merge(DoubleWritable o) {
      return iterate(o);
    }

    public DoubleWritable terminate() {
      return mEmpty ? null : new DoubleWritable(mMax);
    }
  }

  static public class MaxStringevaluator implements UDAFevaluator {
    private Text mMax;
    private boolean mEmpty;

    public MaxStringevaluator() {
      super();
      init();
    }

    public void init() {
      mMax = null;
      mEmpty = true;
    }

    public boolean iterate(Text o) {
      if (o != null) {
        if (mEmpty) {
          mMax = new Text(o);
          mEmpty = false;
        } else if (mMax.compareTo(o) < 0) {
          mMax.set(o);
        }
      }
      return true;
    }

    public Text terminatePartial() {
      return mEmpty ? null : mMax;
    }

    public boolean merge(Text o) {
      return iterate(o);
    }

    public Text terminate() {
      return mEmpty ? null : mMax;
    }
  }

}

public class GenericUDTFExplode extends GenericUDTF {

  private transient ObjectInspector inputOI = null;
  @Override
  public void close() throws HiveException {
  }

  @Override
  public StructObjectInspector initialize(ObjectInspector[] args) throws UDFArgumentException {
    if (args.length != 1) {
      throw new UDFArgumentException("explode() takes only one argument");
    }

    ArrayList fieldNames = new ArrayList();
    ArrayList fieldOIs = new ArrayList();

    switch (args[0].getCategory()) {
    case LIST:
      inputOI = args[0];
      fieldNames.add("col");
      fieldOIs.add(((ListObjectInspector)inputOI).getListElementObjectInspector());
      break;
    case MAP:
      inputOI = args[0];
      fieldNames.add("key");
      fieldNames.add("value");
      fieldOIs.add(((MapObjectInspector)inputOI).getMapKeyObjectInspector());
      fieldOIs.add(((MapObjectInspector)inputOI).getMapValueObjectInspector());
      break;
    default:
      throw new UDFArgumentException("explode() takes an array or a map as a parameter");
    }

    return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames,
        fieldOIs);
  }

  private transient final Object[] forwardListObj = new Object[1];
  private transient final Object[] forwardMapObj = new Object[2];

  @Override
  public void process(Object[] o) throws HiveException {
    switch (inputOI.getCategory()) {
    case LIST:
      ListObjectInspector listOI = (ListObjectInspector)inputOI;
      List list = listOI.getList(o[0]);
      if (list == null) {
        return;
      }
      for (Object r : list) {
        forwardListObj[0] = r;
        forward(forwardListObj);
      }
      break;
    case MAP:
      MapObjectInspector mapOI = (MapObjectInspector)inputOI;
      Map map = mapOI.getMap(o[0]);
      if (map == null) {
        return;
      }
      for (Entry r : map.entrySet()) {
        forwardMapObj[0] = r.getKey();
        forwardMapObj[1] = r.getValue();
        forward(forwardMapObj);
      }
      break;
    default:
      throw new TaskExecutionException("explode() can only operate on an array or a map");
    }
  }

  @Override
  public String toString() {
    return "explode";
  }
}

进击吧大数据

从事大数据行业多年，涉及范围包括不局限于基础支撑、计算引擎、数据整合、数据应用等多方向，参与过大型企业数仓体系建设、对数据建模、数据治理、业务增长有一定的理解;曾收获过多家一线大厂offer，目前带领团队建设企业实时数仓；欢迎大佬们一起加入交流成长