1. 问题描述

收集日志avro数据中有两个Map字段appInstall、appUse分别表示已安装的app、正在使用的app，且key值为app的名称，value值为app使用信息。现在要得到一份匹配上购物类app支付宝|京东|淘宝|天猫的用户名单；MapReduce 解决办法如下：

public static class M extends Mapper
    
      {    Text text = new Text();        @SuppressWarnings("unchecked")    @Override    protected void map(String key, Pair value, Context context) throws IOException, InterruptedException {        Map data = value.fields.data;                String dvc = data.get("dvc").toString();        Map
     
       appInstall = (Map
      
       ) data.get("appInstall");        Map
       
         appUse = (Map
        
         ) data.get("appUse");                for(String app: appInstall.keySet()) {            if(app.matches("支付宝|京东|淘宝|天猫")) {                text.set(appInstall.keySet().toString());                context.write(dvc, text);                return;            }        }                for(String app: appUse.keySet()) {            if(app.matches("支付宝|京东|淘宝|天猫")) {                text.set(appUse.keySet().toString());                context.write(dvc, text);                return;            }        }    }}
        
       
      
     
    

但是，如果要匹配游戏类的app、金融类的app类呢？如果匹配关键词发生了变化呢？显然，我们应该将匹配关键词开放成API，可以自由地匹配正则表达式。这时，pig派上了用场。

2. Bag正则匹配

A = load '/
    
     /
     
      ' using org.apache.pig.piggybank.storage.avro.AvroStorage();-- A: {key: chararray,value: (fields: (data: map[]))}B = foreach A generate value.fields.data#'dvc' as dvc, value.fields.data#'appInstall' as ins:map[], value.fields.data#'appUse' as use:map[];-- B: {dvc: bytearray,ins: map[],use: map[]}C = foreach B generate dvc, KEYSET(ins) as insk, KEYSET(use) as usek;-- C: {dvc: bytearray,insk: {(chararray)},usek: {(chararray)}}
     
    

在上述代码中，load 数据转换得到bag类型的app-set（insk与usek）；但是，应如何遍历bag中的tuple与正则表达式做匹配呢？答案是UDF。

提供了丰富的UDF，其中关于bags的UDF可以参看。 提供根据index从bag提取tuple，支持三个输入参数。依葫芦画瓢，遍历bag匹配正则表达式的UDF如下：

package com.pig.udf.bag;/** * This UDF will return true if one tuple from a bag matches regex. *  *  There are two input parameter: *      1. DataBag *      2. Regex String */public class BagMatchRegex extends FilterFunc {    @Override    public Boolean exec(Tuple tinput) throws IOException {        try{            DataBag samples = (DataBag) tinput.get(0);            String regex = (String) tinput.get(1);            for (Tuple tuple : samples) {                if(((String) tuple.get(0)).matches(regex)){                    return true;                }            }        }        catch (Exception e) {            return false;        }        return false;    }}

其中，FilterFunc为过滤UDF的基类，继承于EvalFunc<Boolean>，即exec(Tuple tinput)的返回值必为Boolean类型。bag正则匹配的pig脚本如下：

REGISTER ../piglib/udf-0.0.1-SNAPSHOT-jar-with-dependencies.jardefine BagMatchRegex com.pig.udf.bag.BagMatchRegex();A = load '/user/../current/*.avro' using org.apache.pig.piggybank.storage.avro.AvroStorage();B = foreach A generate value.fields.data#'dvc' as dvc, value.fields.data#'appInstall' as ins:map[], value.fields.data#'appUse' as use:map[];C = foreach B generate dvc, KEYSET(ins) as insk, KEYSET(use) as usek;D = filter C by BagMatchRegex(insk, '支付宝|京东|淘宝|天猫') or BagMatchRegex(usek, '支付宝|京东|淘宝|天猫');

3. 优化

还有没有可以做优化的地方呢？我们先来看看pig中的实现：

package org.apache.pig.builtin;public class KEYSET extends EvalFunc
    
      {    private static final TupleFactory TUPLE_FACTORY = TupleFactory.getInstance();    @SuppressWarnings("unchecked")    @Override    public DataBag exec(Tuple input) throws IOException {        if(input == null || input.size() == 0) {            return null;        }        Map
     
       m = null;        // Input must be of type Map. This is verified at compile time        m = (Map
      
       )(input.get(0));        if(m == null) {            return null;        }        DataBag bag = new NonSpillableDataBag(m.size());        for (String s : m.keySet()) {            Tuple t = TUPLE_FACTORY.newTuple(s);            bag.add(t);        }        return bag;    }    ...}
      
     
    

需要指出的一点——pig的map数据类型是由Java类Map<String, Object>实现的。从KEYSET源码中可以看出在调用时已经将map遍历了一次，然后在调用BagMatchRegex时又需要将key-set的bag再遍历一次。其实，完全可以只用一次遍历做map-key值的正则匹配：

package com.pig.udf.map;/** * This UDF will return true if map's key matches regex. *  *  There are two input parameter: *      1. Map *      2. Regex String */public class KeyMatchRegex extends FilterFunc {        @SuppressWarnings("unchecked")    @Override    public Boolean exec(Tuple input) throws IOException    {        try{            Map
    
      m = null;            // Input must be of type Map. This is verified at compile time            m = (Map
     
      )(input.get(0));                        String regex = (String) input.get(1);            for (String key : m.keySet()) {                if(key.matches(regex)){                    return true;                }            }        }        catch (Exception e) {            return false;        }        return false;    }}