导读 | Pig是一种数据流语言和运行环境,用于检索非常大的数据集。为大型数据集的处理提供了一个更高层次的抽象。Pig包括两部分:一是用于描述数据流的语言,称为Pig Latin;二是用于运行Pig Latin程序的执行环境。 |
收集日志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派上了用场。
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。
Apache DataFu Pig 提供了丰富的UDF,其中关于bags的UDF可以参看这里。TupleFromBag 提供根据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
REGISTER ../piglib/udf-0.0.1-SNAPSHOT-jar-with-dependencies.jar define 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, '支付宝|京东|淘宝|天猫');
还有没有可以做优化的地方呢?我们先来看看pig中的KEYSET实现:
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
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{ Mapm = 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; } }
原文来自:
本文地址://lrxjmw.cn/pig-keyword-matching.html编辑:吴康宁,审核员:逄增宝
Linux大全:
Linux系统大全: