Apache Pig简介与实践

Apache Pig是一个用来分析大数据集的平台，它由两部分组成：一部分是用于表达数据分析程序的高级脚本语言，另一部分是用于评估分析程序的基本工具。目前来看，Pig主要用于离线数据的批量处理应用场景，但是随着Pig的发展处理数据的速度会不断地提升，这可能依赖于Pig底层的执行引擎。比如，Pig通过指定执行模式，可以使用Hadoop的MapReduce计算引擎来实现数据处理，也可以使用基于Tez的计算引擎来实现（Tez是为了绕开MapReduce多阶段Job写磁盘而设计的DAG计算引擎，性能应该比MapReduce要快），看到Pig未来的发展路线图，以后可能会基于Storm或Spark计算平台实现底层计算引擎，那样速度会有极大地提升。我们基于最新的0.15.0版本的Pig（Hadoop使用的是2.2.0版本），通过编写一些例子脚本来实践Pig的语言特性。

Pig安装与执行

Pig安装非常简单，只需要下载Pig包，然后解压缩即可：

wget http://mirror.bit.edu.cn/apache/pig/pig-0.15.0/pig-0.15.0.tar.gz
tar xvzf pig-0.15.0.tar.gz
sudo ln -s /usr/local/pig-0.15.0 /usr/local/pig
cd /usr/local/pig
bin/pig -x mapreduce

如果希望直接使用pig命令，可以修改环境变量文件~/.bashrc，增加如下配置：

export PIG_HOME=/usr/local/pig
export PATH=$PATH:$PIG_HOME/bin

使变量配置生效：

. ~/.bashrc

Pig支持如下4种执行模式：

• 本地模式

本地模式主要是基于本地文件系统，比较适合调试脚本使用。进入本地模式执行如下命令：

pig -x local

• Tez本地模式

Tez本地模式类似于前面的本地模式，它使用Tez运行时引擎，进入Tez本地模式执行如下命令：

pig -x tez_local

不过该模式还处于试验阶段，不过多累述。

• MapReduce模式

MapReduce模式基于Hadoop，数据存储在HDFS上，它基于运行于YARN之上的MapReduce进行处理。进入MapReduce运行模式执行如下命令：

pig -x mapreduce

一般，我们的数据都是存储在HDFS上的，使用该模式能够充分利用Hadoop集群的计算能力。

• Tez模式

基于Tez模式执行，需要在安装Hadoop集群的时候，修改Hadoop配置文件mapred-site.xml，将属性mapreduce.framework.name的值设置为yarn-tez。进入Tez模式执行如下命令：

pig -x tez

有关Tez相关内容，可以查看Apache Tez官网介绍。

数据类型

基本操作符

• 算数操作符（Arithmetic Operators）包括：+、-、*、/、%、?:、CASE WHEN THEN ELSE END。
• 布尔操作符（Boolean Operators）包括：AND、OR、IN、NOT。
• 类型转换操作符（Cast Operators）：使用圆括号包含类型名，作用于一个字段，例如(int)age、(map[])、(chararray)COUNT($2)、(tuple(chararray,int,map[]))name_age_scores等等。
• 比较操作符（Comparison Operators）包括：==、!=、、=、matches。其中，matches比较操作符使用 Java 的Pattern进行匹配来比较，例如user_name matches ‘[a-n]{3,12}’。
• 类型构造操作符（Type Construction Operators）：可以创建复杂类型的数据，tuple使用()，map使用[]，bag使用{}，例如FOREACH users GENERATE (name, age, address)。
• 解引用操作符（Dereference Operators）：解引用主要是针对集合类型tuple、bag、map，从集合中拿到对应字段的值。比如对于tuple，定义类型t=tuple(t1:int,t2:int,t3:int)，则我要获取字段t1和t3的值，一种方式可以通过t.t1和t.t3得到，也可以通过t.$0和t.$2获取到。

关系操作符

关系操作符示例

• ASSERT

live_user_ids = LOAD '/test/live_user_ids' USING PigStorage() AS (udid: chararray);
ASSERT live_user_ids BY udid != null, 'udid MUST NOT be NULL!';

上面断言表live_user_ids中的udid字段一定存在值。

• GROUP

根据某个或某些字段进行分组，只根据一个字段进行分组，比较简单。如果想要根据两个字段分组，则可以将两个字段构造成一个tuple，然后进行分组。Pig脚本如下所示：

events = LOAD '/data/etl/hive_input/20150613/basis_event_2015061306-r-00002'  USING PigStorage('\t');
projected_events = FOREACH events GENERATE $1 AS (event_code: long), $2 AS (udid: chararray), $10 AS (network: chararray), $34 AS (area_code: int); -- $1是表events的第2个字段
uniq_events = DISTINCT projected_events;
uniq_events = FILTER uniq_events BY (event_code IS NOT NULL) AND (udid IS NOT NULL) AND (network IS NOT NULL) AND (area_code IS NOT NULL);
groupped = GROUP uniq_events BY (udid, event_code) PARALLEL 2; -- 指定使用2个Reducer
selected10 = LIMIT groupped 10;
DUMP selected10;

从HDFS加载的文件中执行投影操作，生成包含event_code、udid、network、area_code这4个字段的一个表projected_events，接着执行去重、条件过滤操作，计算分组的时候基于event_code、udid两个进行分组。

• FOREACH

FOREACH操作可以针对一个数据集进行迭代处理操作，生成一个新的数据集。它有2种使用方法，一种是执行投影操作，选择部分字段的数据，例如脚本：

provinces = LOAD '/test/provinces' USING PigStorage(',') AS (country_id: int, province_id: int, name: chararray);
compositekeyed_provinces = FOREACH provinces GENERATE (CONCAT(CONCAT((chararray)country_id, '_'), (chararray)province_id) AS pid, name);

这里，将原数据集的两个主键字段的值进行拼接合并，作为新表的一个字段。另一种是，支持在FOREACH操作中使用代码段，可以增加更复杂的处理逻辑，摘自官网的例子，例如脚本：

a = LOAD '/test/data' AS (url:chararray, outlink:chararray);
DUMP a;
(www.ccc.com,www.hjk.com)
(www.ddd.com,www.xyz.org)
(www.aaa.com,www.cvn.org)
(www.www.com,www.kpt.net)
(www.www.com,www.xyz.org)
(www.ddd.com,www.xyz.org)
b = GROUP a BY url;
DUMP b;
(www.aaa.com,{(www.aaa.com,www.cvn.org)})
(www.ccc.com,{(www.ccc.com,www.hjk.com)})
(www.ddd.com,{(www.ddd.com,www.xyz.org),(www.ddd.com,www.xyz.org)})
(www.www.com,{(www.www.com,www.kpt.net),(www.www.com,www.xyz.org)})
result = FOREACH b {
	filterda = FILTER a BY outlink == 'www.xyz.org'; -- 过滤掉outlink字段值为'www.xyz.org'的记录
	filtered_outlinks = filterda.outlink;
	filtered_outlinks = DISTINCT filtered_outlinks;  -- 对outlink集合进行去重
	GENERATE group, COUNT(filtered_outlinks); -- 根据对表a进行分组得到group，计算每个分组中outlink的数量
};
DUMP result;
(www.aaa.com,0)
(www.ccc.com,0)
(www.ddd.com,1)
(www.www.com,1)

上面，表b的第一个字段为chararray类型的字段（存放域名字符串），第二个字段是一个bag类型的字段（存在当前域名的出链接，即的集合），最后统计的结果是给定的url的出链接的个数。

• FILTER

根据条件进行过滤，相当于 SQL 中WHERE子句。示例脚本如下所示：

live_info = LOAD '/test/live_info' USING PigStorage() AS (id: long, name: chararray); 
newly_added_lives = FILTER live_info BY (id % 140000 >= 0) AND (name matches '[a-zA-Z0-9]{8, 32}' OR name == 'test');

上面内容很好理解，不再累述。

• JOIN

JOIN操作支持支持配置并行度，指定Reducer的数量。表连接操作，支持内连接和外连接，内连接脚本示例如下：

live = LOAD '/test/shiyj/pig/pig_live' USING PigStorage('\t') AS (id: long, name: chararray);
program = LOAD '/test/shiyj/pig/pig_live_program' USING PigStorage('\t') AS (id: long,name: chararray,live_id: long,live_start: chararray,live_end: chararray);
program_info = JOIN live BY id, program BY live_id;
DUMP program_info;

根据表live的id字段，表program的live_id字段进行内连接。外连接的操作，官网给出了4个例子，可以分别看一下。左外连接例子如下：

A = LOAD 'a.txt' AS (n:chararray, a:int);
B = LOAD 'b.txt' AS (n:chararray, m:chararray);
C = JOIN A by $0 LEFT OUTER, B BY $0; -- 表A和B的字段n进行左外连接

全外连接的例子如下所示：

A = LOAD 'a.txt' AS (n:chararray, a:int);
B = LOAD 'b.txt' AS (n:chararray, m:chararray);
C = JOIN A BY $0 FULL, B BY $0; -- 使用FULL关键字

支持复制的左外连接（Replicated Join），示例如下：

A = LOAD 'large';
B = LOAD 'tiny';
C= JOIN A BY $0 LEFT, B BY $0 USING 'replicated';

只有左外连接支持这种方式，实际上Replicated Join会在MapReduce的Map阶段把做左表进行复制，也就是说做表应该是小表，能够在内存中放得下，然后与右表进行连接操作。还有一种使用Skewed Join，示例如下所示：

A = LOAD 'studenttab' as (name, age, gpa);
B = LOAD 'votertab' as (name, age, registration, contribution);
C = JOIN A BY name FULL, B BY name USING 'skewed';

只有在进行外连接的两表的数据，明显不对称，称为数据倾斜，一个表很大，另一个表相对小，但是内存中放不下，这种情况可以使用Skewed Join操作。目前，Pig支持基于两表的Skewed Join操作。

• DISTINCT

去重操作使用DISTINCT，比较简单，示例如下所示：

live_user_ids = LOAD '/test/live_user_ids' USING PigStorage() AS (udid: chararray);
uniq_user_ids = DISTINCT live_user_ids PARALLEL 8;

DISTINCT操作支持配置并行度，指定Reducer的数量。

• UNION

计算并集操作，使用UNION操作符，示例如下所示：

a = LOAD 'data' AS (a1:int,a2:int,a3:int);
b = LOAD 'data' AS (b1:int,b2:int);
u = UNION a, b;

计算并集，不要求两表的字段数一定相同。

• LIMIT

LIMIT选择计算结果的一部分，示例如下所示：

top100_user_ids = LIMIT live_user_ids 100;

• STORE

STORE操作用来保存计算结果，示例如下所示：

STORE play_users INTO '/test/shiyj/tmp.play_users' USING PigStorage ('\t');

如果没有指定USING子句，则默认使用PigStorage()函数，另外Pig还支持如下的Store/Load函数：

BinStorage()
JsonLoader(['schema'])
JsonStorage()
PigDump()
PigStorage([field_delimiter] , ['options'])
TextLoader()
HBaseStorage('columns', ['options'])
AvroStorage(['schema|record name'], ['options'])
TrevniStorage(['schema|record name'], ['options'])
AccumuloStorage(['columns'[, 'options']])
OrcStorage(['options'])

具体使用方法，可以参考文档介绍。

• CROSS

比较容易理解，摘自官网上的例子，如下所示：

-- 加载表a数据
a = LOAD 'data1' AS (a1:int,a2:int,a3:int);
DUMP a;
(1,2,3)
(4,2,1)

-- 加载表b数据
b = LOAD 'data2' AS (b1:int,b2:int);
DUMP b;
(2,4)
(8,9)
(1,3)

-- 计算笛卡尔积，并输出结果
result = CROSS a, b;
DUMP result;
(1,2,3,2,4)
(1,2,3,8,9)
(1,2,3,1,3)
(4,2,1,2,4)
(4,2,1,8,9)
(4,2,1,1,3)

• CUBE

这个操作符功能比较强大，如下所示：

users = LOAD '/test/shiyj/pig/pig_live_users' AS (create_date,room_id,audio_id,udid);
groupped = COGROUP users BY (create_date,room_id,audio_id);
groupped_count = FOREACH groupped {
	uniq = DISTINCT users.udid;
	GENERATE group, COUNT(uniq);
};
STORE groupped_count INTO '/test/shiyj/pig/groupped_count' USING PigStorage('\t'); -- 将分组统计的结果保存到HDFS
groupped_count = LOAD '/test/shiyj/pig/groupped_count/part-r-*' AS (k: tuple(chararray, long, long), cnt: int); -- 加载前面保存的结果，进行CUBE计算
groupped_count = FOREACH groupped_count GENERATE k.$0, k.$1, k.$2, cnt;
cubed_users = CUBE groupped_count BY CUBE($0, $1, $2);
DUMP cubed_users;

salesinp = LOAD '/pig/data/salesdata' USING PigStorage(',') AS (product:chararray, year:int, region:chararray, state:chararray, city:chararray, sales:long);
cubedinp = CUBE salesinp BY CUBE(product,year);
result = FOREACH cubedinp GENERATE FLATTEN(group), SUM(cube.sales) AS totalsales;
DUMP result;

如果输入数据为(car, 2012, midwest, ohio, columbus, 4000)，则上面脚本执行CUBE操作，结果输出内容如下所示：

(car,2012,4000)
(car,,4000)
(,2012,4000)
(,,4000)

salesinp = LOAD '/pig/data/salesdata' USING PigStorage(',') AS (product:chararray, year:int, region:chararray, state:chararray, city:chararray, sales:long);
rolledup = CUBE salesinp BY ROLLUP(region,state,city);
result = FOREACH rolledup GENERATE FLATTEN(group), SUM(cube.sales) AS totalsales;
DUMP result;

同样如果输入tuple值为(car, 2012, midwest, ohio, columbus, 4000)，则ROLLUP操作结果如下所示：

(midwest,ohio,columbus,4000)
(midwest,ohio,,4000)
(midwest,,,4000)
(,,,4000)

salesinp = LOAD '/pig/data/salesdata' USING PigStorage(',') AS (product:chararray, year:int, region:chararray, state:chararray, city:chararray, sales:long);
cubed_and_rolled = CUBE salesinp BY CUBE(product,year), ROLLUP(region, state, city); 
result = FOREACH cubed_and_rolled GENERATE FLATTEN(group), SUM(cube.sales) AS totalsales;

上面的CUBE操作等价于下面两中操作：

cubed_and_rolled = CUBE salesinp BY CUBE(product,year,region, state, city); 
-- 或
cubed_and_rolled = CUBE salesinp BY ROLLUP(product,year,region, state, city);

执行结果，如下所示：

(car,2012,midwest,ohio,columbus,4000)
(car,2012,midwest,ohio,,4000)
(car,2012,midwest,,,4000)
(car,2012,,,,4000)
(car,,midwest,ohio,columbus,4000)
(car,,midwest,ohio,,4000)
(car,,midwest,,,4000)
(car,,,,,4000)
(,2012,midwest,ohio,columbus,4000)
(,2012,midwest,ohio,,4000)
(,2012,midwest,,,4000)
(,2012,,,,4000)
(,,midwest,ohio,columbus,4000)
(,,midwest,ohio,,4000)
(,,midwest,,,4000)
(,,,,,4000)

• SAMPLE

对数据进行取样操作，脚本如下所示：

-- 加载原始数据集，并计算记录数
users = LOAD '/test/shiyj/pig/pig_live_users' AS (create_date,room_id,audio_id,udid);
g_users = GROUP users ALL;
total_user_cnt = FOREACH g_users GENERATE COUNT(users);
DUMP total_user_cnt;

-- 15%取样，计算取样记录数
sampled_users = SAMPLE users 0.15;
g_sampled_users = GROUP sampled_users ALL;
sampled_user_cnt = FOREACH g_sampled_users GENERATE COUNT(sampled_users);
DUMP sampled_user_cnt;

• MAPREDUCE

MAPREDUCE操作允许在Pig脚本内部执行MapReduce程序，示例脚本来自官网，如下所示：

A = LOAD 'WordcountInput.txt';
B = MAPREDUCE 'wordcount.jar' STORE A INTO 'inputDir' LOAD 'outputDir'
    AS (word:chararray, count: int) `org.myorg.WordCount inputDir outputDir`;

如果直接写原生的MapReduce程序各个能解决实际问题，可以将写好的程序打包，在Pig脚本中指定相关参数即可运行。

• SPLIT

将一个表拆分成多个表，可以按照“水平拆分”的思想进行操作，根据某些条件来生成新表。示例如下所示：

A = LOAD 'data' AS (f1:int,f2:int,f3:int); -- 加载数据到表A
DUMP A;               
(1,2,3)
(4,5,6)
(7,8,9)       

SPLIT A INTO X IF f1<7, Y IF f2==5, Z IF (f3<6 OR f3>6); -- A表中满足条件f1<7的记录插入到表X中，满足条件f2==5的记录插入到T表中，满足条件(f3<6 OR f3>6)的记录插入到Z表中

DUMP X;
(1,2,3)
(4,5,6)

DUMP Y;
(4,5,6)

DUMP Z;
(1,2,3)
(7,8,9)

求值函数（Eval Functions）

a = LOAD 'a' as (x, y); 
b = LOAD 'b' as (x, y);
c = JOIN a by x, b by x; -- 表a和b连接，因为表a和b有相同的列名，所以连接后添加前缀“表名::”来区分

DEFINE pluck PluckTuple('a::'); <code>-- 定义前缀"a::"，等价于DEFINE pluck PluckTuple('a::', true);
d = FOREACH c GENERATE FLATTEN(pluck(*)); -- 包含前缀"a::"的都保留
DESCRIBE c;
c: {a::x: bytearray,a::y: bytearray,b::x: bytearray,b::y: bytearray}
DESCRIBE d;
d: {plucked::a::x: bytearray,plucked::a::y: bytearray}

DEFINE pluckNegative PluckTuple('a::','false'); -- 定义前缀"a::"，包含该前缀的过滤掉
d = FOREACH c GENERATE FLATTEN(pluckNegative(*));
DESCRIBE d; -- 结果中包含前缀"a::"的被排除掉
d: {plucked::b::x: bytearray,plucked::b::y: bytearray}

数学函数

数学函数比较简单，不再详细描述，主要包括如下20个：

ABS
ACOS
ASIN
ATAN
CBRT
CEIL
COS
COSH
EXP
FLOOR
LOG
LOG10
RANDOM
ROUND
ROUND_TO
SIN
SINH
SQRT
TAN
TANH

具体使用可以查看官方文档。

字符串函数

字符串函数非常常用，主要包括如下20个：

ENDSWITH
EqualsIgnoreCase
INDEXOF
LAST_INDEX_OF
LCFIRST
LOWER
LTRIM
REGEX_EXTRACT
REGEX_EXTRACT_ALL
REPLACE
RTRIM
SPRINTF
STARTSWITH
STRSPLIT
STRSPLITTOBAG
SUBSTRING
TRIM
UCFIRST
UPPER
UniqueID

使用方法可以查看文档。

日期时间函数

日期函数有下面24个，如下所示：

AddDuration
CurrentTime
DaysBetween
GetDay
GetHour
GetMilliSecond
GetMinute
GetMonth
GetSecond
GetWeek
GetWeekYear
GetYear
HoursBetween
MilliSecondsBetween
MinutesBetween
MonthsBetween
SecondsBetween
SubtractDuration
ToDate
ToMilliSeconds
ToString
ToUnixTime
WeeksBetween
YearsBetween

集合函数

Hive UDF函数

在Pig中可以直接调用Hive的UDF，HiveUDAF和HiveUDTF，语法如下表所示：

DEFINE sin HiveUDF('sin'); -- 定义HiveUDF，后面可以直接使用函数sin
a = LOAD 'student' AS (name:chararray, age:int, gpa:double);
b = FOREACH a GENERATE sin(gpa); -- 使用函数sin

DEFINE explode HiveUDTF('explode');
a = LOAD 'mydata' AS (a0:{(b0:chararray)});
b = FOREACH a GENERATE FLATTEN(explode(a0));

DEFINE avg HiveUDAF('avg');
a = LOAD 'student' AS (name:chararray, age:int, gpa:double);
b = GROUP a BY name;
c = FOREACH b GENERATE group, AVG(a.age);

Pig UDF

Pig也支持用户自定义函数UDF，而且支持使用多种编程语言来实现UDF，目前支持的变成语言包括：Java、JavaScript、Jython、 Ruby 、Groovy、Python。以Java为例，可以通过继承自类org.apache.pig.EvalFunc来实现一个UDF，将实现的UDF泪打包后，Pig安装目录下的CLASSPATH下面，然后可以在Pig脚本中使用。例如，我们实现的UDF类为org.shirdrn.pig.udf.IPAddressConverterUDF，用来根据ip代码（long类型），转换为对应的点分十进制的IP地址字符串，打包后JAR文件名称为iptool.jar，则可以在Pig脚本中这样使用：

REGISTER 'iptool.jar';
a = LOAD '/data/etl/$date_string/$event_file' AS (event_code: long, udid: chararray, ip_code: long, network: chararray);
b = FOREACH a GENERATE event_code, udid, org.shirdrn.etl.pig.udf.IPAddressConverterUDF(ip_code);
DUMP b;

相关问题总结

• 运行Pig脚本，从外部向脚本传递参数

在实际使用中，我们经常需要从外部传递参数到Pig脚本中，例如，文件路径，或者日期时间，等等，可以直接使用pig命令的-p参数从外部传参，例如，有下面的Pig脚本compute_event_user_count.pig：

a = LOAD '/data/etl/hive_input/20150613/basis_event_2015061305-r-00001' USING PigStorage('\t');
DESCRIBE a;
b = FOREACH a GENERATE $1 AS event_code, $2 AS udid, $4 AS install_id;
c = GROUP b BY (event_code, udid);
d = FOREACH c GENERATE group, COUNT($1);
DUMP d;

上面我们是直接将文件路径写死在脚本中，如果需要从外部传递输入文件、输出目录，则可以改写为：

a = LOAD '$input_file' USING PigStorage('\t');
DESCRIBE a;
b = FOREACH a GENERATE $1 AS event_code, $2 AS udid, $4 AS install_id;
c = GROUP b BY (event_code, udid);
d = FOREACH c GENERATE group, COUNT($1);
STORE d INTO '$output_dir' USING PigStorage ('\t');

则可以执行Pig脚本，并传递参数：

bin/pig -p input_file=/data/etl/hive_input/20150613/basis_event_2015061305-r-00001 -p output_dir=/test/shiyj/pig/example_output -x mapreduce compute_event_user_count.pig

这样就可以实现从外部向Pig脚本传递参数，可以到HDFS上查看结果输出文件/test/shiyj/pig/example_output/part-r-00000。pig命令更多选项，可以查看pig帮助命令：

bin/pig -h

• 运行Pig脚本出现异常“Retrying connect to server: 0.0.0.0/0.0.0.0:10020”

实际应用中，我们几乎不可能将Pig安装到Hadoop集群的NameNode所在的节点，如果可以安装到NameNode节点上，基本不会报这个错误的。这个错误主要有是由于Pig没有安装在NameNode节点上，而是以外的其它节点上，它在执行计算过程中，需要与MapReduce的JobHistoryServer的IPC服务进行通信，所以在安装Hadoop时需要允许JobHistoryServer的IPC主机和端口被外部其它节点访问，只需要修改etc/hadoop/mapreduce-site.xml配置文件，增加如下配置即可：

<property>
                <name>mapreduce.jobhistory.address</name>
                <value>10.10.4.130:10020</value>
                <description>MapReduce JobHistory Server IPC host:port</description>
        </property>

如果第一次安装Hadoop没有配置该属性mapreduce.jobhistory.address，那么Hadoop集群的所有节点上会使用默认的配置值为0.0.0.0:10020，所以如果不在NameNode上安装Pig程序，导致Pig所在的节点上安装的Hadoop的配置属性mapreduce.jobhistory.address使用默认值，也就是Pig所在节点0.0.0.0:10020，所以Pig脚本就会执行过程中与本机的10020端口通信，显然会失败的。其实，如果已经在NameNode上启动了JobHistoryServer进程，只需要修改mapreduce.jobhistory.address的属性值，然后同步到所有安装Hadoop文件的节点，包括Pig所在节点即可，不需要重启NameNode节点上的JobHistoryServer进程。如果没有在NameNode上启动JobHistoryServer进程，执行如下命令启动即可：

mr-jobhistory-daemon.sh start historyserver

参考链接