您可以捐助,支持我们的公益事业。

1元 10元 50元





认证码:  验证码,看不清楚?请点击刷新验证码 必填



  求知 文章 文库 Lib 视频 iPerson 课程 认证 咨询 工具 讲座 Modeler   Code  
会员   
 
   
 
 
     
   
 订阅
  捐助
【Pig源码分析】谈谈Pig的数据模型
 
作者:周 明耀 来源:IBM 发布于 2016-2-26
  2511  次浏览      
 

1. 数据模型

Schema

Pig Latin表达式操作的是relation,FILTER、FOREACH、GROUP、SPLIT等关系操作符所操作的relation就是bag,bag为tuple的集合,tuple为有序的field列表集合,而field表示数据块(A field is a piece of data),可理解为数据字段。

Schema为数据所遵从的类型格式,包括:field的名称及类型(names and types)。用户常用as语句来自定义schema,或是load函数导入schema,比如:

A = foreach X generate .. as field1:chararray, .. as field2:bag{};
A = load '..' using PigStorage('\t', '-schema');
A = load '..' using org.apache.pig.piggybank.storage.avro.AvroStorage();

若不指定field的类型,则其默认为bytearray。对未知schema进行操作时,有:

若join/cogroup/cross多关系操作遇到未知schema,则会将其视为null schema,导致返回结果的schema也为null;

若flatten一个empty inner schema的bag(即:bag{})时,则返回结果的schema为null;

若union时二者relation的schema不一致,则返回结果的schema为null;

若field的schema为null,会将该字段视为bytearray。

为了保证pig脚本运行的有效性,在写UDF时要在outputSchema方法中指定返回结果的schema。

数据类型

Pig的基本数据类型与对应的Java类:

复杂数据类型及其对应的Java类:

Pig的复杂数据类型可以嵌套表达,比如:tuple中有tuple (a, (b, c, d)),tuple中有bag (a, {(b,c), (d,e)})等等。但是一定要遵从数据类型本身的定义,比如:bag中只能是tuple的集合,比如{a, {(b),(c)}}就是不合法的。

Pig还有一种特殊的数据类型:null,与Java、C中null不一样,其表示不知道的或不存在的数据类型(unknown or non-existent)。比如,在load数据时,如果有的数据行字段不符合定义的schema,则该字段会被置为null。

2. 源码分析

以下源码分析采用的是0.12版本。

Tuple

在KEYSET源码中,创建Tuple对象采用工厂+单例设计模式:

private static final TupleFactory TUPLE_FACTORY = TupleFactory.getInstance();
Tuple t = TUPLE_FACTORY.newTuple(s);

事实上,TupleFactory是个抽象类,实现接口TupleMaker<Tuple>。在方法TupleFactory.getInstance()中,默认情况下返回的是BinSedesTupleFactory对象,同时支持加载用户重写的TupleFactory类(pig.data.tuple.factory.name指定类名、 pig.data.tuple.factory.jar指定类所在的jar)。BinSedesTupleFactory继承于TupleFactory:

在BinSedesTupleFactory的newTuple方法中,返回的是BinSedesTuple对象。BinSedesTuple类继承于DefaultTuple类,在DefaultTuple类中有List<Object> mFields字段,这便是存储Tuple数据的地方了,mFields所持有类型为ArrayList<Object>();。类图关系:

Bag

创建Bag对象有下面几种方法:

// factory
BagFactory mBagFactory = BagFactory.getInstance();
DataBag output = mBagFactory.newDefaultBag();

// if you know upfront how many tuples you are going to put in this bag.
DataBag bag = new NonSpillableDataBag(m.size());

与TupleFactory一样,BagFactory也是抽象类,也支持用户自定义重写;getInstance方法默认返回的是DefaultBagFactory

DefaultBagFactory有newDefaultBag、newSortedBag、newDistinctBag方法分别创建三类bag:

default bag中的tuple没有排序,也没有去重;

sorted bag中的tuple是按序存放,顺序是由tuple default comparator或bag创建时的comparator所定义的;

distinct bag顾名思义,tuple有去重。

三类bag的构造器如下:

public DefaultDataBag() {
mContents = new ArrayList<Tuple>();
}

public SortedDataBag(Comparator<Tuple> comp) {
mComp = (comp == null) ? new DefaultComparator() : comp;
mContents = new ArrayList<Tuple>();
}

public DistinctDataBag() {
mContents = new HashSet<Tuple>();
}

BagFactory的类图:

DefaultAbstractBag作为三种类型bag的基类,有一个字段mContents用于存放tuple,NonSpillableDataBag直接实现DataBag接口。DataBag的类图:

3. 实战

现有avro日志数据(见前一篇),其字段:

1.dvc表示用户手机标识;

2.appUse与appInstall同为avro Map类型,其key为app名称(app name),value为Map<String, Object>,包含了一个表示使用时间的字段timelist(类型为ArrayList);具体格式如下

'dvc': 'imei_123',
'appUse': {
'app name1': {
...
'timelist': [...]
},
'app name2': {
...
'timelist': [...]
},
...
},
'appInstall': {
'app name1': {
...
'timelist': [...]
},
...
}

现在,想要得到每个用户的app列表及app的打开次数,以格式dvc, {(app)}, {(app, frequency)}输出,即用户 + app列表 + 使用次数类表。如果用MapRduce做,得分为以下步骤:

1.以(dvc, app)为key值,计算value值为使用次数;

2.以dvc为key值,合并同一用户的不同app,value值为(app, fre);

3.以dvc为key值,计算appinstall的app列表;

4.将步骤2得到的数据与步骤3得到的数据做join,然后输出。

可以看出用MapReduce略显繁复,如何来用pig来实现呢?我们可以对appUse:map[]编写EVAL UDF,让其返回(app名称, timelist的长度) :

public class AppTimelist  extends EvalFunc<DataBag>{
private static final TupleFactory TUPLE_FACTORY = TupleFactory.getInstance();
private static final BagFactory BAG_FACTORY = BagFactory.getInstance();

@SuppressWarnings({ "unchecked" })
@Override
public DataBag exec(Tuple input) throws IOException {
Map<String, Map<String, Object>> m = (Map<String, Map<String, Object>>) input.get(0);
List<Object> result = new ArrayList<Object>();
DataBag output = BAG_FACTORY.newDefaultBag();

if(m == null)
return null;

for(Map.Entry<String, Map<String, Object>> e: m.entrySet()) {
result.clear();
String app = e.getKey();
long size = ((DataBag) e.getValue().get("timelist")).size();
result.add(app);
result.add(size);
output.add(TUPLE_FACTORY.newTuple(result));
}

return output;
}
}

pig将Java的ArrayList转成DataBag的类型,所以要对timelist进行强转操作。

对appInstall:map[]编写EVAL UDF,返回(appList):

public class DistinctBag extends EvalFunc<DataBag> {
BagFactory mBagFactory = BagFactory.getInstance();

@Override
public DataBag exec(Tuple input) throws IOException {
if(input == null || input.size() == 0) {
return null;
}

DataBag in = (DataBag) input.get(0);
DataBag out = mBagFactory.newDistinctBag();

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

for(Tuple tp: in) {
DataBag applist = (DataBag) tp.get(0);
for(Tuple app: applist)
out.add(app);
}
return out;
}
}

上面提到过,若没有给EVAL UDF指定返回值的schema,则返回结果的schema为null,如此会造成类型的丢失,在后面的操作中容易报NullPointerException。

// AppTimelist.java
@Override
public Schema outputSchema(Schema input) {
try {
Schema tupleSchema = new Schema();
FieldSchema chararrayFieldSchema = new Schema.FieldSchema(null, DataType.CHARARRAY);
FieldSchema longFieldSchema = new Schema.FieldSchema(null, DataType.LONG);
tupleSchema.add(chararrayFieldSchema);
tupleSchema.add(longFieldSchema);
return new Schema(new Schema.FieldSchema(getSchemaName(this
.getClass().getName().toLowerCase(), input), tupleSchema,
DataType.TUPLE));
} catch (Exception e) {
return null;
}
}

// DistinctBag.java
@Override
public Schema outputSchema(Schema input) {
FieldSchema innerFieldSchema = new Schema.FieldSchema(null, DataType.CHARARRAY);
Schema innerSchema = new Schema(innerFieldSchema);
Schema bagSchema = null;

try {
bagSchema = new Schema(new FieldSchema(null, innerSchema, DataType.BAG));
} catch(FrontendException e) {
throw new RuntimeException(e);
}
return bagSchema;
}

统计app列表:

define AvroStorage org.apache.pig.piggybank.storage.avro.AvroStorage;
define DistinctBag com.pig.udf.bag.DistinctBag;
A = load '..' using AvroStorage();
B = foreach A generate value.fields.data#'dvc' as dvc:chararray, value.fields.data#'appInstall' as ins:map[map[]];
C = foreach B generate dvc, KEYSET(ins) as applist;
D = group C by dvc;
-- extract applist from grouped D
E = foreach D {
projected = foreach $1 generate applist;
generate group as dvc, projected as grouped;
}
F = foreach E generate dvc, DistinctBag(grouped) as applist;
store F into '..' using AvroStorage();

统计app使用时长:

define AvroStorage org.apache.pig.piggybank.storage.avro.AvroStorage;
define AppTimelist com.pig.udf.map.AppTimelist;
A = load '..' using AvroStorage();
B = foreach A generate value.fields.data#'dvc' as dvc:chararray, value.fields.data#'appUse' as use:map[map[]];
C = foreach B generate dvc, flatten(AppTimelist(use)) as (app, fre);
D = group C by (dvc, app);
E = foreach D generate flatten(group) as (dvc, app), SUM($1.fre) as fre;
F = group E by dvc;
G = foreach F {
projected = foreach $1 generate app, fre;
generate group as dvc, projected as appfre;
}
store G into '..' using AvroStorage();

二者做join即可得到结果。

   
2511 次浏览       
相关文章

基于EA的数据库建模
数据流建模(EA指南)
“数据湖”:概念、特征、架构与案例
在线商城数据库系统设计 思路+效果
 
相关文档

Greenplum数据库基础培训
MySQL5.1性能优化方案
某电商数据中台架构实践
MySQL高扩展架构设计
相关课程

数据治理、数据架构及数据标准
MongoDB实战课程
并发、大容量、高性能数据库设计与优化
PostgreSQL数据库实战培训
最新课程计划

MySQL索引背后的数据结构
MySQL性能调优与架构设计
SQL Server数据库备份与恢复
让数据库飞起来 10大DB2优化
oracle的临时表空间写满磁盘
数据库的跨平台设计
更多...   

并发、大容量、高性能数据库
高级数据库架构设计师
Hadoop原理与实践
Oracle 数据仓库
数据仓库和数据挖掘
Oracle数据库开发与管理

GE 区块链技术与实现培训
航天科工某子公司 Nodejs高级应用开发
中盛益华 卓越管理者必须具备的五项能力
某信息技术公司 Python培训
某博彩IT系统厂商 易用性测试与评估
中国邮储银行 测试成熟度模型集成(TMMI)
中物院 产品经理与产品管理
更多...