pig 笔记

朴灿烈づ我的快乐病毒、 2022-06-04 05:13 203阅读 0赞

Pig是一个基于[Hadoop][]的大规模[数据分析][Link 1]平台，它提供的SQL-LIKE语言叫Pig Latin，该语言的编译器会把类SQL的数据分析请求转换为一系列经过优化处理的[MapReduce][]运算。Pig为复杂的[海量数据][Link 2][并行计算][Link 3]提供了一个简单的操作和编程接口。

![Center][]

![Center 1][]

![Center 2][]

![Center 3][]

![Center 4][]

![Center 5][]

![Center 6][]

![Center 7][]

map：key 是char array ；value 可以任意类型。

\[’name’\#’bob’,’age’\#55\]:创建一个包含 name 和 age 两个键的map。

tuple：是一个定长，无须相同类型。

（’bob’,55）:创建一个包含两个字段的tuple 常量。

bag：是一个无序的tuple集合。无须加载到内存中的数据类型。

\{(‘bob’,55),(’sally’,52),(‘john,25’)\}：构造了一个包含的3个 tuple 的 bag，每个tuple 包含两个字段。

包是元祖的集合，里面的元素可以不是元祖。relation是特殊的包，里面的元素是元祖。

![Center 8][]

![Center 9][]

**pig关系操作符：**

**![Center 10][]**

**![Center 11][]关系名上面有个student**

**![Center 12][]**

**还需要dump g；**

**![Center 13][]**

**![Center 14][]**

**产生结果 ：one is age,another is a bag;**

**![Center 15][]**

**![Center 16][]**

**![Center 17][]**

**对多个关系进行分组，合并之后的分组。**

**![Center 18][]**

**![Center 19][]**

**![Center 20][]**

**![Center 21][]**

**![Center 22][]**

**![Center 23][]**

**![Center 24][]**

**![Center 25][]**

**![Center 26][]**

**![Center 27][]**

**![Center 28][]**

**----------------------------------------------------------------------------------------------------------------------------------------------------------**

所有命令和脚本都在Pig 0.12.0 & Hadoop 2.2.0下测试通过。

准备两个数据文件：

1）student.txt 结构为（班级号，学号，成绩）,字段间逗号分隔。

C01,N0101,82

C01,N0102,59

C01,N0103,65

C02,N0201,81

C02,N0202,82

C02,N0203,79

C03,N0301,56

C03,N0302,92

C03,N0306,72

2）teacher.txt 结构为（班级号，教师）,字段间逗号分隔。

C01,Zhang

C02,Sun

C03,Wang

C04,Dong

# 加载和存储（Load，Store） #

执行以下命令

records = load'hdfs://localhost:9000/input/student.txt' using PigStorage(',') as(classNo:chararray, studNo:chararray, score:int);

dump records;

store records into ' hdfs://localhost:9000/input/student\_out' using PigStorage(':');

然后查看hdfs://localhost:9000/input/student\_out目录下的part-m-00000文件，其内容如下：

C01:N0101:82

C01:N0102:59

C01:N0103:65

C02:N0201:81

C02:N0202:82

C02:N0203:79

C03:N0301:56

C03:N0302:92

C03:N0306:72

其中的load是加载操作，store是存储操作。他们分别可以指定其分隔符，比如上例中的逗号和分号。

# 筛选（Filter） #

执行以下命令：

records\_c01 = filter records byclassNo=='C01';

dump records\_c01;

结果如下：

(C01,N0101,82)

(C01,N0102,59)

(C01,N0103,65)

注意：判断是否相等要用两个等号。

# Foreach Generate #

Foreach对关系中的每一个记录循环，然后按指定模式生成一个新的关系。

执行以下命令：

score\_c01 = foreach records\_c01generate 'Teacher',$1,score;

dump score\_c01;

结果如下：

(Teacher,N0101,82)

(Teacher,N0102,59)

(Teacher,N0103,65)

生成的新的关系中包括三个字段，第一个字段是常量，第二个字段是学号（我们是通过索引号引用的），第三个字段是分数（我们通过字段名引用的）。

# 分组（group） #

执行以下命令：

grouped\_records = group recordsby classNo parallel 2;

dump grouped\_records;

结果如下：

(C02,\{(C02,N0203,79),(C02,N0202,82),(C02,N0201,81)\})

(C01,\{(C01,N0103,65),(C01,N0102,59),(C01,N0101,82)\})

(C03,\{(C03,N0306,72),(C03,N0302,92),(C03,N0301,56)\})

其中的Paraller 2表示启用2个Reduce操作。

如何统计每个班级及格和优秀的学生人数呢？执行以下两个命令：

result = foreach grouped\_records \{

fail =filter records by score < 60;

excellent =filter records by score >=90;

generategroup, COUNT(fail) as fail, COUNT(excellent) as excellent;

\};

dump result;

结果如下：

(C01,1,0)

(C02,0,0)

(C03,1,1)

题外话：

flatten操作，可以将数据格式扁平化。我们分别通过tuple和bag来看看flatten的作用：

1）  Flatten对tuple的作用

执行以下命令:

a= foreach records generate $0,($1,$2);

dumpa;

输出结果如下：

(C01,(N0101,82))

(C01,(N0102,59))

(C01,(N0103,65))

(C02,(N0201,81))

(C02,(N0202,82))

(C02,(N0203,79))

(C03,(N0301,56))

(C03,(N0302,92))

(C03,(N0306,72))

然后，执行:

b = foreach a generate $0,flatten($1);

dump b;

结果如下：

(C01,N0101,82)

(C01,N0102,59)

(C01,N0103,65)

(C02,N0201,81)

(C02,N0202,82)

(C02,N0203,79)

(C03,N0301,56)

(C03,N0302,92)

(C03,N0306,72)

由此看见，flatten作用于tuple时，将flatten对应的字段（tuple）中的字段扁平化为关系中的字段。（不知道该如何解释比较好）

2）  Flatten对bag的作用

执行以下命令

c = foreach records generate $0,\{($1),($1,$2)\};

dump c;

结果如下：

(C01,\{(N0101),(N0101,82)\})

(C01,\{(N0102),(N0102,59)\})

(C01,\{(N0103),(N0103,65)\})

(C02,\{(N0201),(N0201,81)\})

(C02,\{(N0202),(N0202,82)\})

(C02,\{(N0203),(N0203,79)\})

(C03,\{(N0301),(N0301,56)\})

(C03,\{(N0302),(N0302,92)\})

(C03,\{(N0306),(N0306,72)\})

接下来执行：

d = foreach c generate $0,flatten($1);

dump d;

结果如下：

(C01,N0101)

(C01,N0101,82)

(C01,N0102)

(C01,N0102,59)

(C01,N0103)

(C01,N0103,65)

(C02,N0201)

(C02,N0201,81)

(C02,N0202)

(C02,N0202,82)

(C02,N0203)

(C02,N0203,79)

(C03,N0301)

(C03,N0301,56)

(C03,N0302)

(C03,N0302,92)

(C03,N0306)

(C03,N0306,72)

可以看出，flatten作用于bag时，会消除嵌套关系，生成类似于笛卡尔乘积的结果。（不好表达，读者可以细细体会）。

# Stream操作 #

可以将Python程序嵌入到Pig中使用。

建立一个Python文件pass.py，内容如下：

\#! /usr/bin/envpython

import sys

for line insys.stdin:

(c,n,s) = line.split()

if int(s) >= 60:

print "%s\\t%s\\t%s"%(c,n,s)

执行以下命令:

define pass \`pass.py\` SHIP('/home/user/pass.py');

records\_pass = stream records through pass as(classNo:chararray, studNo:chararray, score:int);

dump records\_pass;

结果如下：

(C01,N0101,82)

(C01,N0103,65)

(C02,N0201,81)

(C02,N0202,82)

(C02,N0203,79)

(C03,N0302,92)

(C03,N0306,72)

可以看出，统计结果为所有及格的记录（>=60）。

其中，ship用于将python程序提交到Hadoop集群中去。

请注意第一个命令中的\`pass.py\`，不是用单引号括起来的，是用键盘1左边的那个键上的字符括起来的。（不知道这个字符怎么称呼，只知道是一种标注符号）

# Join #

先执行以下两条命令：

r\_student = load'hdfs://localhost:9000/input/student.txt' using PigStorage(',') as (classNo:chararray, studNo: chararray, score: int);

r\_teacher2 = load'hdfs://localhost:9000/input/teacher.txt' using PigStorage(',') as (classNo:chararray, teacher: chararray);

回到本文开头，我们有两个数据文件，分别为学生（班级，学号，成绩）；老师（班级，姓名）。

执行以下命令：

r\_joined = join r\_student by classNo,r\_teacher by classNo;

dump r\_joined;

(C01,N0103,65,C01,Zhang)

(C01,N0102,59,C01,Zhang)

(C01,N0101,82,C01,Zhang)

(C02,N0203,79,C02,Sun)

(C02,N0202,82,C02,Sun)

(C02,N0201,81,C02,Sun)

(C03,N0306,72,C03,Wang)

(C03,N0302,92,C03,Wang)

(C03,N0301,56,C03,Wang)

类似于SQL中的内连接Inner Join。当然你也可以使用外连接，比如：

r\_joined = join r\_student by classNo left outer,r\_teacher by classNo;

dump r\_joined;

注意：left outer/right outer要写在第一个关系名的后面。以下语法是错误的：

r\_joined = join r\_student by classNo, r\_teacher by classNo leftouter; //错误

# COGROUP #

Join的操作结果是平面的（一组元组），而COGROUP的结果是有嵌套结构的。

运行以下命令：

r1 = cogroup r\_student by classNo,r\_teacher by classNo;

dump r1;

结果如下：

(C01,\{(C01,N0103,65),(C01,N0102,59),(C01,N0101,82)\},\{(C01,Zhang)\})

(C02,\{(C02,N0203,79),(C02,N0202,82),(C02,N0201,81)\},\{(C02,Sun)\})

(C03,\{(C03,N0306,72),(C03,N0302,92),(C03,N0301,56)\},\{(C03,Wang)\})

(C04,\{\},\{(C04,Dong)\})

由结果可以看出：

1）  cogroup和join操作类似。

2）  生成的关系有3个字段。第一个字段为连接字段；第二个字段是一个包，值为关系1中的满足匹配关系的所有元组；第三个字段也是一个包，值为关系2中的满足匹配关系的所有元组。

3）  类似于Join的外连接。比如结果中的第四个记录，第二个字段值为空包，因为关系1中没有满足条件的记录。实际上第一条语句和以下语句等同：

r1= cogroup r\_student by classNo outer,r\_teacher by classNo outer;

如果你希望关系1或2中没有匹配记录时不在结果中出现，则可以分别在关系中使用inner而关键字进行排除。

执行以下语句：

r1 = cogroup r\_student by classNo inner,r\_teacher byclassNo outer;

dump r1;

结果为：

(C01,\{(C01,N0103,65),(C01,N0102,59),(C01,N0101,82)\},\{(C01,Zhang)\})

(C02,\{(C02,N0203,79),(C02,N0202,82),(C02,N0201,81)\},\{(C02,Sun)\})

(C03,\{(C03,N0306,72),(C03,N0302,92),(C03,N0301,56)\},\{(C03,Wang)\})

如先前我们讲到的flatten，执行以下命令：

r2 = foreach r1 generate flatten($1),flatten($2);

dump r2;

结果如下：

(C01,N0103,65,C01,Zhang)

(C01,N0102,59,C01,Zhang)

(C01,N0101,82,C01,Zhang)

(C02,N0203,79,C02,Sun)

(C02,N0202,82,C02,Sun)

(C02,N0201,81,C02,Sun)

(C03,N0306,72,C03,Wang)

(C03,N0302,92,C03,Wang)

(C03,N0301,56,C03,Wang)

# Cross #

执行以下命令：

r = cross r\_student,r\_teacher;

dump r;

结果如下：

(C03,N0306,72,C04,Dong)

(C03,N0306,72,C03,Wang)

(C03,N0306,72,C02,Sun)

(C03,N0306,72,C01,Zhang)

(C03,N0302,92,C04,Dong)

(C03,N0302,92,C03,Wang)

(C03,N0302,92,C02,Sun)

(C03,N0302,92,C01,Zhang)

(C03,N0301,56,C04,Dong)

(C03,N0301,56,C03,Wang)

(C03,N0301,56,C02,Sun)

(C03,N0301,56,C01,Zhang)

(C02,N0203,79,C04,Dong)

(C02,N0203,79,C03,Wang)

(C02,N0203,79,C02,Sun)

(C02,N0203,79,C01,Zhang)

(C02,N0202,82,C04,Dong)

(C02,N0202,82,C03,Wang)

(C02,N0202,82,C02,Sun)

(C02,N0202,82,C01,Zhang)

(C02,N0201,81,C04,Dong)

(C02,N0201,81,C03,Wang)

(C02,N0201,81,C02,Sun)

(C02,N0201,81,C01,Zhang)

(C01,N0103,65,C04,Dong)

(C01,N0103,65,C03,Wang)

(C01,N0103,65,C02,Sun)

(C01,N0103,65,C01,Zhang)

(C01,N0102,59,C04,Dong)

(C01,N0102,59,C03,Wang)

(C01,N0102,59,C02,Sun)

(C01,N0102,59,C01,Zhang)

(C01,N0101,82,C04,Dong)

(C01,N0101,82,C03,Wang)

(C01,N0101,82,C02,Sun)

(C01,N0101,82,C01,Zhang)

由此可以看出，cross类似于笛卡尔乘积。一般情况下不建议直接使用cross，而应该事前对数据集进行筛选，提高效率。

# 排序（Order） #

执行以下命令：

r = order r\_student by score desc, classNo asc;

dump r;

结果如下：

(C03,N0302,92)

(C01,N0101,82)

(C02,N0202,82)

(C02,N0201,81)

(C02,N0203,79)

(C03,N0306,72)

(C01,N0103,65)

(C01,N0102,59)

(C03,N0301,56)

# 联合（Union） #

执行以下语句：

r\_union = union r\_student, r\_teacher;

dump r\_union;

结果如下：

(C01,N0101,82)

(C01,N0102,59)

(C01,N0103,65)

(C02,N0201,81)

(C02,N0202,82)

(C02,N0203,79)

(C03,N0301,56)

(C03,N0302,92)

(C03,N0306,72)

(C01,Zhang)

(C02,Sun)

(C03,Wang)

(C04,Dong)

可以看出：

1）  union是取两个记录集的并集。

2）  关系r\_union的schema为未知（unknown），这是因为被union的两个关系的schema是不一样的。如果两个关系的schema是一致的，则union后的关系将和被union的关系的schema一致。

**----------------------------------------------------------------------------------------------------------------------------------------------------------**

从实例出发

%default file test.txt

A = load '$file' as (date, web, name, food);

B = load '$file' as (date, web, name, food);

C= cogroup A by $0, B by $1;

describe C;

illustrate C;

dump C;

cogroup命令中$0和$1，两个列的内容如果不一样，就是分别生成两个批次的group，先按A值分组，在按B对应的值分组。按A的值分组时，B对应的为空，则group中有一个空组\{\}；但如果内容一样，如C= cogroup A by $1, B by $1;就是生成一个批次的group，其中包含A和B两个表中所有的等于该值的元组。

COGROUP与join的区别：自己懒得写，摘自网络

Join的操作结果是平面的（一组元组），而COGROUP的结果是有嵌套结构的。  
运行以下命令：  
r1 = cogroup r\_student by classNo,r\_teacher by classNo;  
dump r1;  
结果如下：  
(C01,\{(C01,N0103,65),(C01,N0102,59),(C01,N0101,82)\},\{(C01,Zhang)\})  
(C02,\{(C02,N0203,79),(C02,N0202,82),(C02,N0201,81)\},\{(C02,Sun)\})  
(C03,\{(C03,N0306,72),(C03,N0302,92),(C03,N0301,56)\},\{(C03,Wang)\})  
(C04,\{\},\{(C04,Dong)\})  
由结果可以看出：  
1）  cogroup和join操作类似。  
2）  生成的关系有3个字段。第一个字段为连接字段；第二个字段是一个包，值为关系1中的满足匹配关系的所有元组；第三个字段也是一个包，值为关系2中的满足匹配关系的所有元组。  
3）  类似于Join的外连接。比如结果中的第四个记录，第二个字段值为空包，因为关系1中没有满足条件的记录。实际上第一条语句和以下语句等同：  
r1= cogroup r\_student by classNo outer,r\_teacher by classNo outer;  
如果你希望关系1或2中没有匹配记录时不在结果中出现，则可以分别在关系中使用inner而关键字进行排除。  
执行以下语句：  
r1 = cogroup r\_student by classNo inner,r\_teacher byclassNo outer;  
dump r1;  
结果为：  
(C01,\{(C01,N0103,65),(C01,N0102,59),(C01,N0101,82)\},\{(C01,Zhang)\})  
(C02,\{(C02,N0203,79),(C02,N0202,82),(C02,N0201,81)\},\{(C02,Sun)\})  
(C03,\{(C03,N0306,72),(C03,N0302,92),(C03,N0301,56)\},\{(C03,Wang)\})

flatten执行命令：  
r2 = foreach r1 generate flatten($1),flatten($2);  
dump r2;  
结果如下：  
(C01,N0103,65,C01,Zhang)  
(C01,N0102,59,C01,Zhang)  
(C01,N0101,82,C01,Zhang)  
(C02,N0203,79,C02,Sun)  
(C02,N0202,82,C02,Sun)  
(C02,N0201,81,C02,Sun)  
(C03,N0306,72,C03,Wang)  
(C03,N0302,92,C03,Wang)  
(C03,N0301,56,C03,Wang)

可以看到，两个同时flatten，会自动映射生成多列。

针对cogroup，我测试了一下，核心代码如下：

industry\_existed\_Data =  LOAD '$industryPath' USING PigStorage(',') AS (industryId:chararray,guid:chararray,sex:chararray,log\_type:chararray);  
sample\_data = limit industry\_existed\_Data 20;  
\--STORE  sample\_data INTO '/user/wizad/tmp/industry\_existed\_Data' USING PigStorage(',');  
\--merge with history data   
cogroupIndustryExistCurrentByGuid = COGROUP industry\_existed\_Data by guid, industry\_current\_data by guid;  
mydata = sample cogroupIndustryExistCurrentByGuid 0.1;  
dump mydata;  
describe cogroupIndustryExistCurrentByGuid;  
\--dump cogroupIndustryExistCurrentByGuid;  
  
\--STORE  mycogroupdata INTO '/user/wizad/tmp/cogroupIndustryExistCurrentByGuid' USING PigStorage(',');  
  
look\_for\_cogroup = FOREACH cogroupIndustryExistCurrentByGuid GENERATE $0,$2;  
describe look\_for\_cogroup;  
  
IndustryStorageDataTmp = FOREACH cogroupIndustryExistCurrentByGuid GENERATE FLATTEN($2);  
IndustryStorageData = DISTINCT IndustryStorageDataTmp;  
describe IndustryStorageData;

显示结果：

三个数据的结构如下

cogroupIndustryExistCurrentByGuid:   
\{  
group: chararray,  
industry\_existed\_Data:\{industryId: chararray,guid: chararray,sex: chararray,log\_type: chararray\},  
industry\_current\_data: \{joined\_ad\_campaign\_data::industryId: chararray,joined\_Orgin\_sex\_data::distinct\_origin\_historical\_sex::guid: chararray,joined\_Orgin\_sex\_data::social\_sex::sex: chararray,joined\_Orgin\_sex\_data::distinct\_origin\_historical\_sex::log\_type: chararray\}  
\}  
look\_for\_cogroup:   
\{  
group: chararray,  
industry\_current\_data: \{joined\_ad\_campaign\_data::industryId: chararray,joined\_Orgin\_sex\_data::distinct\_origin\_historical\_sex::guid: chararray,joined\_Orgin\_sex\_data::social\_sex::sex: chararray,joined\_Orgin\_sex\_data::distinct\_origin\_historical\_sex::log\_type: chararray\}  
\}  
IndustryStorageData:   
\{  
industry\_current\_data::joined\_ad\_campaign\_data::industryId: chararray,  
industry\_current\_data::joined\_Orgin\_sex\_data::distinct\_origin\_historical\_sex::guid: chararray,  
industry\_current\_data::joined\_Orgin\_sex\_data::social\_sex::sex: chararray,  
industry\_current\_data::joined\_Orgin\_sex\_data::distinct\_origin\_historical\_sex::log\_type: chararray  
\}

可以看出三个数据的结构很复杂，因为前面做关联所以包含了对象名（或者叫域名），指明属于哪个对象。可以只看最后一列名字和格式。

第三个是flatten（$2）的结果。

cogroup有空集问题，就是对应group中的每个值（cogroup用来关联的key的取值），两个集合各自按key值进行group后，某些key对应的集合为空。

上面的pig代码的实际数据如下，guid作为关联key，可以看出很多空集\{\}，出现在某些guid的取值对应集合后。

所以取数据时要注意，只flatten某一列，会造成其他列数据丢失，因为对应着该flatten列的空集。

((-1,),\{(74,9051235c-a391-4dae-ab22-f93d24a12636,-1,-1,),(75,053e9f48-03bf-4b39-9455-ff412a725a3c,-1,-1,),(74,21ca723c-ec2b-4242-8108-b95436f10e3e,-1,-1,),(74,fec1932a-b0e4-4bf0-b504-8ed8f3c159e7,-1,-1,),(74,d74374ec-8cf4-4c4a-b598-9631f6972cbb,-1,-1,),(74,6780962a-bf75-4c4c-a557-94a7de5a3e36,-1,-1,),(74,14517915-ee3f-4d34-943f-d6f1813afdef,-1,-1,),(74,c5547aca-3b8b-4108-93ba-bf365c106cdd,-1,-1,),(74,e9a986c1-6868-4f7f-baf6-69d8c302583e,-1,-1,),(74,9c1341cf-45b8-48c6-b699-33b1a4215c66,-1,-1,),(74,f16e6222-a84b-4758-ae71-0613c8f34b29,-1,-1,),(74,47cc25ef-05bc-47f4-a32b-3cddaf0ac22b,-1,-1,),(74,d5c1b6b0-38c3-464b-8cb9-70ced875be5f,-1,-1,),(74,6a4f782a-1f5c-45c0-bb3a-4df25c436be3,-1,-1,),(74,23bb2f0c-d629-479d-800e-b86fc3d6e45c,-1,-1,)\})  
((a50a17bde79ac018,),\{(74,863010025134441,a50a17bde79ac018,863010025134441,)\})  
((a51779f736cd3f54,),\{(74,862949029595753,a51779f736cd3f54,862949029595753,)\})  
((c7ae5867-3b77-4987-b082-ed3867b5c384,),\{(74,353627055387065,c7ae5867-3b77-4987-b082-ed3867b5c384,353627055387065,)\})

新建了一个文件test.txt,随便造了几条测试数据(tab键分隔，是pig默认的分隔方式，若要以其他分隔，load的时候load 'test.txt' using PigStorage(',') as .... 就可以以逗号分隔)：

xiaojun 28      上海  
yangna  24      兰州

在pig grant shell中输入中文是没法解析的，会报错，比如在shell下面输入:

a = load 'test.txt' as (name:chararray,age:int,city:chararray);

b = filter a by city == '上海'; //此句执行会报错。

dump b;

如果需要使用到中文进行数据集的过滤，可以将语句写到一个单独的pig脚本中,例如我们新建一个test.pig文件，把上面两句写入到该文件中去,然后再执行pig -x local test.pig,这时可以成功的执行中文字符的过滤语句。

一般情况下，都会有定期运行的pig脚本，比如每天运行一次的脚本，这类脚本往往在内部都要用到当天的日期作为参数，pig支持参数替换，参数由前缀$字符来标示，例如我们需要load每天的数据进行统计分析，每天的数据在hdfs上是按日期进行进行命名的，我们可以新建一个test.pig脚本：

a = load '$input' as (.....);

store a into '$output';

然后我们在命令行可以这样：pig -param input=/user/tom/input/2014-12-01.dat -param output=/user/tom/output/2014-12-01.dat -f test.pig

test.pig中会把$标示符指定的参数替换为-param指定的参数值。

更常用的是通过动态参数来替换，即shell脚本中经常会使用到的反引号引用的命令。

我们可以把2014-12-01这段通过shell脚本动态输出：

pig -param input=/user/tom/input/\`date "+%Y-%m-%d"\`.dat -param output=/user/tom/output/\`date "+%Y-%m-%d"\`.dat -f test.pig

这样就实现了参数的完全动态替换。

假设文件test4.txt有这么两行数据：

1980080113312121212018  
1985080113313131313023

规则是前8位为年月日，中间11位为手机号码，后3位表示的是年龄。

我们可以自定义一个加载udf来加载这个文件

\[java\]  [view plain][]  [copy][view plain]

1.  package com.besttone.pig.udf.load;  
2.    
3.  import java.io.IOException;  
4.  import java.util.List;  
5.    
6.  import org.apache.commons.logging.Log;  
7.  import org.apache.commons.logging.LogFactory;  
8.  import org.apache.hadoop.io.Text;  
9.  import org.apache.hadoop.mapreduce.InputFormat;  
10. import org.apache.hadoop.mapreduce.Job;  
11. import org.apache.hadoop.mapreduce.RecordReader;  
12. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;  
13. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;  
14. import org.apache.pig.LoadFunc;  
15. import org.apache.pig.backend.executionengine.ExecException;  
16. import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit;  
17. import org.apache.pig.data.DataByteArray;  
18. import org.apache.pig.data.Tuple;  
19. import org.apache.pig.data.TupleFactory;  
20.   
21. public class CutLoadFunc extends LoadFunc \{  
22.   
23.     protected static final Log log = LogFactory.getLog(CutLoadFunc.class);  
24.     private final List<Range> ranges;  
25.   
26.     private RecordReader reader;  
27.   
28.     private final TupleFactory tupleFactory = TupleFactory.getInstance();  
29.   
30.     public CutLoadFunc(String cutPattern) throws IOException \{  
31.         ranges = Range.parse(cutPattern);  
32.     \}  
33.   
34.     @Override  
35.     public void setLocation(String location, Job job) throws IOException \{  
36.         // TODO Auto-generated method stub  
37.         FileInputFormat.setInputPaths(job, location);  
38.     \}  
39.   
40.     @Override  
41.     public InputFormat getInputFormat() throws IOException \{  
42.         // TODO Auto-generated method stub  
43.         return new TextInputFormat();  
44.     \}  
45.   
46.     @Override  
47.     public void prepareToRead(RecordReader reader, PigSplit split)  
48.             throws IOException \{  
49.         // TODO Auto-generated method stub  
50.         this.reader = reader;  
51.     \}  
52.   
53.     @Override  
54.     public Tuple getNext() throws IOException \{  
55.         // TODO Auto-generated method stub  
56.   
57.         try \{  
58.             if (!reader.nextKeyValue()) \{  
59.                 return null;  
60.             \}  
61.             Text value = (Text) reader.getCurrentValue();  
62.             String line = value.toString();  
63.   
64.             Tuple tuple = tupleFactory.newTuple(ranges.size());  
65.             for (int i = 0; i < ranges.size(); i++) \{  
66.                 Range range = ranges.get(i);  
67.                 if (range.getEnd() > line.length()) \{  
68.                     log.warn(String.format(  
69.                             "Range end (%s) is longer than line length (%s)",  
70.                             range.getEnd(), line.length()));  
71.                     continue;  
72.                 \}  
73.                 tuple.set(i, new DataByteArray(range.getSubstring(line)));  
74.   
75.             \}  
76.             return tuple;  
77.   
78.         \} catch (InterruptedException e) \{  
79.             throw new ExecException(e);  
80.         \}  
81.   
82.     \}  
83.   
84. \}

\[java\]  [view plain][]  [copy][view plain]

1.  package com.besttone.pig.udf.load;  
2.    
3.  import java.io.IOException;  
4.  import java.util.ArrayList;  
5.  import java.util.List;  
6.    
7.  import org.apache.commons.logging.Log;  
8.  import org.apache.commons.logging.LogFactory;  
9.    
10. public class Range \{  
11.   
12.     protected static final Log log = LogFactory.getLog(Range.class);  
13.     private int start;  
14.     private int end;  
15.   
16.     public int getStart() \{  
17.         return start;  
18.     \}  
19.   
20.     public void setStart(int start) \{  
21.         this.start = start;  
22.     \}  
23.   
24.     public int getEnd() \{  
25.         return end;  
26.     \}  
27.   
28.     public void setEnd(int end) \{  
29.         this.end = end;  
30.     \}  
31.   
32.     public static List<Range> parse(String cutPattern) throws IOException \{  
33.   
34.         String\[\] rangelist = cutPattern.split(",");  
35.         if (rangelist.length == 0) \{  
36.             throw new IOException("cutPattern参数不合法");  
37.         \}  
38.         List<Range> list = new ArrayList<Range>(rangelist.length);  
39.         for (String rangestr : rangelist) \{  
40.             try \{  
41.                 Range range = new Range();  
42.                 range.setStart(Integer.parseInt(rangestr.split("-")\[0\]));  
43.                 range.setEnd(Integer.parseInt(rangestr.split("-")\[1\]));  
44.                 list.add(range);  
45.             \} catch (Exception e) \{  
46.                 throw new IOException("cutPattern参数不合法");  
47.             \}  
48.         \}  
49.         return list;  
50.   
51.     \}  
52.   
53.     public String getSubstring(String line) \{  
54.         if (this.end > line.length()) \{  
55.             log.warn(String.format(  
56.                     "Range end (%s) is longer than line length (%s)", this.end,  
57.                     line.length()));  
58.             return null;  
59.         \}  
60.         return line.substring(this.start, this.end);  
61.     \}  
62.       
63.     public static void main(String\[\] args) \{  
64.         Range range = new Range();  
65.         range.setStart(19);  
66.         range.setEnd(22);  
67.         System.out.println(range.getSubstring("1980080113312121212018"));  
68.     \}  
69. \}

将这两个类打成jar包,然后进入pig grunt,执行一下脚本

a= load 'test4.txt' using com.besttone.pig.udf.load.CutLoadFunc('0-8,8-19,19-22') as (date:chararray,phone:chararray,age:chararray);

dump a;

可以看到内容被成功的加载：

(19800801,13312121212,018)  
(19850801,13313131313,023)

[Hadoop]: https://baike.baidu.com/item/Hadoop
[Link 1]: https://baike.baidu.com/item/%E6%95%B0%E6%8D%AE%E5%88%86%E6%9E%90/6577123
[MapReduce]: https://baike.baidu.com/item/MapReduce/133425
[Link 2]: https://baike.baidu.com/item/%E6%B5%B7%E9%87%8F%E6%95%B0%E6%8D%AE/2271586
[Link 3]: https://baike.baidu.com/item/%E5%B9%B6%E8%A1%8C%E8%AE%A1%E7%AE%97/113443
[Center]: /images/20220604/728e0bd536cb4d15b5eaf2b8f5c281c6.png
[Center 1]: /images/20220604/6d68d6029a154a4194fd18174941847e.png
[Center 2]: /images/20220604/d135c22fa2464964a5bab15b0162529c.png
[Center 3]: /images/20220604/190c368eec104c9eaa4fa65597e9b8da.png
[Center 4]: /images/20220604/2d1e493f0fd1463096d5a5f3145b1d23.png
[Center 5]: /images/20220604/1516b50d7a984f8f8ce8a8ae75e87077.png
[Center 6]: /images/20220604/1a112193b16041ec93de98be3cde78d5.png
[Center 7]: /images/20220604/17492a3f944b4de39a01ad82f968b846.png
[Center 8]: /images/20220604/a66b282ed8f5423fb88552a197217a79.png
[Center 9]: /images/20220604/368a100403584bb38996551f932e8a91.png
[Center 10]: /images/20220604/e402d5f8bcc444a3aad7e002b6fe1dc0.png
[Center 11]: /images/20220604/43490b63aedf40debb765f28ca89c19c.png
[Center 12]: /images/20220604/a74dc9f8eff94237811fd0bf1f541cd4.png
[Center 13]: /images/20220604/567ba45a5be94115bbec291cfa2fcc78.png
[Center 14]: /images/20220604/8209a7021375465aaee074f381fe3490.png
[Center 15]: /images/20220604/a34ff3e50dcf4ffe85e5ac0bbc4053d1.png
[Center 16]: /images/20220604/9e4986e49b6c4754a19159073f21ce4a.png
[Center 17]: /images/20220604/7b828b527c4343cda0026d40d79ebd9d.png
[Center 18]: /images/20220604/2a8ee00091754c6aa71ce49e8032985f.png
[Center 19]: /images/20220604/a13e9794e1274fd6bf1c15613f68519b.png
[Center 20]: /images/20220604/d89ecb34934543c9aa78253b96856c3a.png
[Center 21]: /images/20220604/7581e895c2ae4501b7457515ef84f7ae.png
[Center 22]: /images/20220604/06549c13e81645b8bbf82ac0ae66a868.png
[Center 23]: /images/20220604/030de62b730546d78fe45b57eb9ccd59.png
[Center 24]: /images/20220604/32488ee9dafd47f49f5aa224e7f32b7c.png
[Center 25]: /images/20220604/8d38b56ccacf4a0ebfc6606afa4b6ec1.png
[Center 26]: /images/20220604/0d6e685f7e164ec09ee2b4339c53df51.png
[Center 27]: /images/20220604/ebc4466aaf5d44b69408d5cc28fa5c07.png
[Center 28]: /images/20220604/5247a489cb374c98a3a6143b832402ad.png
[view plain]: http://blog.csdn.net/xiao_jun_0820/article/details/41893311#