[原创]Apache Pig中文教程（进阶）

本文包含Apache Pig的一些进阶技巧及用法小结。如要学习基础教程，请查看我写的【其他几篇文章】。
本文的大量实例都是作者Darran Zhang（website: codelast.com）在工作、学习中总结的经验或解决的问题，并且添加了较为详尽的说明及注解，此外，作者还在不断地添加本文的内容，希望能帮助一部分人。

Apache Pig是用来处理大规模数据的高级查询语言，配合Hadoop使用，可以在处理海量数据时达到事半功倍的效果，比使用Java，C++等语言编写大规模数据处理程序的难度要小N倍，实现同样的效果的代码量也小N倍。

本文基于以下环境：
pig 0.8.1
文章来源：http://www.codelast.com/
✔ 如何编写及使用自定义函数（UDF）
首先给出一个链接：Pig 0.8.1 API，还有Pig UDF Manual。这两个文档能提供很多有用的参考。
自定义函数有何用？这里以一个极其简单的例子来说明一下。
假设你有如下数据：

[root@localhost pig]$ cat a.txt 
uidk    12  3
hfd    132 99
bbN    463 231
UFD    13  10

现在你要将第二列的值先+500，再-300，然后再÷2.6，那么我们可以这样写：

grunt> A = LOAD 'a.txt' AS(col1:chararray, col2:double, col3:int);
grunt> B = FOREACH A GENERATE col1, (col2 + 500 - 300)/2.6, col3;
grunt> DUMP B;
(uidk,81.53846153846153,3)
(hfd,127.6923076923077,99)
(bbN,255.0,231)
(UFD,81.92307692307692,10)

我们看到，对第二列进行了 (col2 + 500 - 300)/2.6 这样的计算。麻烦不？或许这点小意思没什么。但是，如果有比这复杂得多的处理，每次你需要输入多少pig代码呢？我们希望有这样一个函数，可以让第二行pig代码简化如下：

grunt> B = FOREACH A GENERATE col1, com.codelast.MyUDF(col2), col3;

这样的话，对于我们经常使用的操作，岂不是很方便？
pig的UDF（user-defined function）就是拿来做这个的。
文章来源：http://www.codelast.com/
下面，就以IntelliJ这个IDE为例（其实用什么IDE倒无所谓，大同小异吧），说明我们如何实现这样一个功能。
①新建一个新工程，在工程下创建“lib”目录，然后把pig安装包中的“pig-0.8.1-core.jar”文件放置到此lib目录下，然后在“Project Structure→Libraries”下添加（点击“+”号）一个库，就命名为“lib”，然后点击右侧的“Attach Classes”按钮，选择pig-0.8.1-core.jar文件，再点击下方的“Apply”按钮应用此更改。这样做之后，你就可以在IDE的编辑器中实现输入代码时看到智能提示了。
此外，你还需要用同样的方法，将一堆Hadoop的jar包添加到工程中，包括以下文件：

hadoop-XXX-ant.jar
hadoop-XXX-core.jar
hadoop-XXX-examples.jar
hadoop-XXX-test.jar
hadoop-XXX-tools.jar

其中，XXX是版本号。
如果没有这些文件，你在编译jar包的时候会报错。
文章来源：http://www.codelast.com/
②跟我一起，在工程目录下的 src/com/coldelast/ 目录下创建Java源代码文件 MyUDF.java，其内容如下：

package com.codelast;

import java.io.IOException;

import org.apache.pig.EvalFunc;
import org.apache.pig.data.Tuple;

/**
 * Author: Darran Zhang @ codelast.com
 * Date: 2011-09-29
 */

public class MyUDF extends EvalFunc<Double> {

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

    try {
      Double val = (Double) input.get(0);
      val = (val + 500 - 300) / 2.6;
      return val;
    } catch (Exception e) {
      throw new IOException(e.getMessage());
    }
  }
}

在上面的代码中，input.get(0)是获取UDF的第一个参数（可以向UDF传入多个参数）；同理，如果你的UDF接受两个参数（例如一个求和的UDF），那么input.get(1)可以取到第二个参数。
然后编写build.xml（相当于C++里面的Makefile），用ant来编译、打包此工程——这里就不把冗长的build.xml写上来了，而且这也不是关键，没有太多意义。
文章来源：http://www.codelast.com/
③假定编译、打包得到的jar包名为cl.jar，我们到这里几乎已经完成了大部分工作。下面就看看如何在pig中调用我们刚编写的自定义函数了。

grunt> REGISTER cl.jar;
grunt> A = LOAD 'a.txt' AS(col1:chararray, col2:double, col3:int);    
grunt> B = FOREACH A GENERATE col1, com.codelast.MyUDF(col2), col3;
grunt> DUMP B;
(uidk,81.53846153846153,3)
(hfd,127.6923076923077,99)
(bbN,255.0,231)
(UFD,81.92307692307692,10)

注：第一句是注册你编写的UDF，使用前必须先注册。
从结果可见，我们实现了预定的效果。
UDF大有用途！
注意：对如果你的UDF返回一个标量类型（类似于我上面的例子），那么pig就可以使用反射（reflection）来识别出返回类型。如果你的UDF返回的是一个包（bag）或一个元组（tuple），并且你希望pig能理解包（bag）或元组（tuple）的内容的话，那么你就要实现outputSchema方法，否则后果很不好。具体可看这个链接的说明。

✔ 怎样自己写一个UDF中的加载函数(load function)
①加载函数(load function)是干什么的？
先举一个很简单的例子，来说明load function的作用。
假设有如下数据：

[root@localhost pig]# cat a.txt
1,2,3
a,b,c
9,5,7

我们知道，pig默认是以tab作为分隔符来加载数据的，所以，如果你没有指定分隔符的话，将使得每一行都被认为只有一个字段：

grunt> B = FOREACH A GENERATE $0;
grunt> DUMP B;
(1,2,3)
(a,b,c)
(9,5,7)

而我们想要以逗号作为分隔符，则应该使用pig内置函数PigStorage：

A = LOAD 'a.txt' using PigStorage(',');

这样的话，我们再用上面的方法DUMP B，得到的结果就是：

(1)
(a)
(9)

这个例子实在太简单了，在这里，PigStorage这个函数就是一个加载函数(load function)。
定义：

Load/Store Functions

 

These user-defined functions control how data goes into Pig and comes out of Pig. Often, the same function handles both input and output but that does not have to be the case.

即：加载函数定义了数据如何流入和流出pig。一般来说，同一函数即处理输入数据，又处理输出数据，但并不是必须要这样。
有了这个定义，就很好理解加载函数的作用了。再举个例子：你在磁盘上保存了只有你自己知道怎么读取其格式的数据（例如，数据是按一定规则加密过的，只有你知道如何解密成明文），那么，你想用pig来处理这些数据，把它们转换成一个个字段的明文时，你就必须要有这样一个加载函数(load function)，来进行LOAD数据时的转换工作。这就是加载函数(load function)的作用。
文章来源：http://www.codelast.com/
②知道了load function是干嘛的，现在怎么写一个load function？如果你看的是这个链接的UDF手册：Pig Wiki UDF Manual中，会发现它是这样说的——
加载函数必须要实现 LoadFunc 接口，这个接口类似于下面的样子：

public interface LoadFunc {
    public void bindTo(String fileName, BufferedPositionedInputStream is, long offset, long end) throws IOException;
    public Tuple getNext() throws IOException;
    // conversion functions
    public Integer bytesToInteger(byte[] b) throws IOException;
    public Long bytesToLong(byte[] b) throws IOException;
    ......
    public void fieldsToRead(Schema schema);
    public Schema determineSchema(String fileName, ExecType execType, DataStorage storage) throws IOException;
}

其中：

• bindTo函数在pig任务开始处理数据之前被调用一次，它试图将函数与输入数据关联起来。
• getNext函数读取输入的数据流并构造下一个元组（tuple）。当完成数据处理时该函数会返回null，当该函数无法处理输入的元组（tuple）时它会抛出一个IOException异常。
• 接下来就是一批转换函数，例如bytesToInteger，bytesToLong等。这些函数的作用是将数据从bytearray转换成要求的类型。
• fieldsToRead函数被保留作未来使用，应被留空。
• determineSchema函数对不同的loader应有不同的实现：对返回真实数据类型（而不是返回bytearray字段）的loader，必须要实现该函数；其他类型的loader只要将determineSchema函数返回null就可以了。

但是，如果你在IDE中import了pig 0.8.1的jar包“pig-0.8.1-core.jar”，会发现 LoadFunc 根本不是一个接口（interface），而是一个抽象类（abstract class），并且要实现的函数也与该文档中所说的不一致。因此，只能说是文档过时了。
所以，要看文档的话，还是要看这个Pig UDF Manual，这里面的内容才是对的。
同时，我也推荐另外一个关于Load/Store Function的链接：《Programming Pig》Chapter 11. Writing Load and Store Functions。这本书很好很强大。

③开始写一个loader。我们现在写一个①中所描述的、可以按逗号分隔符加载数据文件的loader——PigStorage已经有这个功能了，不过为了演示loader是怎么写出来的，这里还是用这个功能来说明。
代码如下：

package com.codelast.udf.pig;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.pig.*;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.*;
import org.apache.pig.data.*;

import java.io.IOException;
import java.util.*;

/**
 * A loader class of pig.
 *
 * @author Darran Zhang (codelast.com)
 * @version 11-10-11
 * @declaration These codes are only for non-commercial use, and are distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND, either express or implied.
 * You must not remove this declaration at any time.
 */

public class MyLoader extends LoadFunc {
  protected RecordReader recordReader = null;

  @Override
  public void setLocation(String s, Job job) throws IOException {
    FileInputFormat.setInputPaths(job, s);
  }

  @Override
  public InputFormat getInputFormat() throws IOException {
    return new PigTextInputFormat();
  }

  @Override
  public void prepareToRead(RecordReader recordReader, PigSplit pigSplit) throws IOException {
    this.recordReader = recordReader;
  }

  @Override
  public Tuple getNext() throws IOException {
    try {
      boolean flag = recordReader.nextKeyValue();
      if (!flag) {
        return null;
      }
      Text value = (Text) recordReader.getCurrentValue();
      String[] strArray = value.toString().split(",");
      List lst = new ArrayList<String>();
      int i = 0;
      for (String singleItem : strArray) {
        lst.add(i++, singleItem);
      }
      return TupleFactory.getInstance().newTuple(lst);
    } catch (InterruptedException e) {
      throw new ExecException("Read data error", PigException.REMOTE_ENVIRONMENT, e);
    }
  }
}

如上，你的loader类要继承自LoadFunc虚类，并且需要重写它的4个方法。其中，getNext方法是读取数据的方法，它做了读取出一行数据、按逗号分割字符串、构造一个元组（tuple）并返回的事情。这样我们就实现了按逗号分隔符加载数据的loader。
文章来源：http://www.codelast.com/
④关于load function不得不说的一些话题
如果你要加载一个数据文件，例如：

A = LOAD 'myfile' AS (col1:chararray, col2:int);

假设此文件的结构不复杂，你可以手工写 AS 语句，但如果此文件结构特别复杂，你总不可能每次都手工写上几十个甚至上百个字段名及类型定义吧？
这个时候，如果我们可以让pig从哪里读出来要加载的数据的schema（模式），就显得特别重要了。
在实现load function的时候，我们是通过实现 LoadMetadata 这个接口中的 getSchema 方法来做到这一点的。例如：

public class MyLoadFunc extends LoadFunc implements LoadMetadata {

  public ResourceSchema getSchema(String filename, Job job) throws IOException {
    //TODO:
  }
}

实现了 getSchema 方法之后，在pig脚本中加载数据的时候，就可以无需编写 AS 语句，就可以使用你在 getSchema 方法中指定的模式了。例如：

REGISTER 'myUDF.jar';
A = LOAD 'myfile' USING com.codelast.MyLoadFunc();
B = foreach A generate col1;
SOTRE B INTO 'output';

看清楚了，在 LOAD 的时候，我们并没有写 AS 语句来指定字段名，但是在后面的 FOREACH 中，我们却可以使用 col1 这样的字段名，这正是因为 getSchema 方法的实现为我们做到了这一点。在数据文件的结构特别复杂的时候，这个功能几乎是不可或缺的，否则难以想像会给分析数据的人带来多大的不便。

✔ 重载(overloading)一个UDF
类似于C++的函数重载，pig中也可以重载UDF，例如一个函数ADD可以对两个int进行操作，也可以对两个double进行操作，那么我们可以为该函数实现 getArgToFuncMapping 方法，该函数返回一个 List<FuncSpec> 对象，这个对象中包含了参数的类型信息。具体怎么实现，可以看这个链接（搜索“Overloading UDFs”定位到所需章节）。

✔ pig运行不起来，提示“org.apache.hadoop.ipc.Client - Retrying connect to server: localhost/127.0.0.1:9000. Already tried 1 time(s)”错误的解决办法
发生这个错误时，请先检查Hadoop的各个进程是否都运行起来了，例如，在我的一次操作中，遇到这个错误时，我发现Hadoop namenode进程没有启动起来：

ps -ef | grep java | grep NameNode

应该有两个进程启动起来了：

org.apache.hadoop.hdfs.server.namenode.NameNode
org.apache.hadoop.hdfs.server.namenode.SecondaryNameNode

如果没有，那么你要到Hadoop安装目录下的“logs”目录下，查看NameNode的日志记录文件（视用户不同，日志文件的名字也会有不同），例如，我的NameNone日志文件 hadoop--namenode-root-XXX.log 的末尾，显示出有如下错误：

ERROR org.apache.hadoop.hdfs.server.namenode.NameNode: org.apache.hadoop.hdfs.server.common.InconsistentFSStateException: Directory /tmp/hadoop-root/dfs/name is in an inconsistent state: storage directory does not exist or is not accessible.

文章来源：http://www.codelast.com/
我到它提示的地方一看，果然不存在最后一级目录（我是伪分布式运行的Hadoop，不要觉得奇怪），于是手工创建了这个目录，然后停掉Hadoop：

stop-all.sh

稍候一会儿再重新启动Hadoop：

start-all.sh

然后再去看一下NameNode的日志，又发现了新的错误信息：

ERROR org.apache.hadoop.hdfs.server.namenode.NameNode: java.io.IOException: NameNode is not formatted.

这表明NameNode没有被格式化。于是将其格式化：

[root@localhost bin]# hadoop namenode -format

命令行问你是否要格式化的时候，选择YES即可。格式化完后会提示：

common.Storage: Storage directory /tmp/hadoop-root/dfs/name has been successfully formatted.

说明成功了。这个时候，再像前面一样重启Hadoop进程，再去看NameNode的日志文件的最后若干行，应该会发现前面的那些错误提示没了。这个时候，再检查Hadoop各进程是否都成功地启动了，如果是的话，那么这个时候你就可以在Hadoop的伪分布式模式下启动pig：

[root@localhost home]# pig

而不用以本地模式来运行pig了（pig -x local）。
总之，配置一个伪分布式的Hadoop来调试pig在某些情况下是很有用的，但是比较麻烦，因为还牵涉到Hadoop的正确配置，但是最好搞定它，以后大有用处啊。

✔ 用Pig加载HBase数据时遇到的错误“ERROR 2999: Unexpected internal error. could not instantiate 'com.twitter.elephantbird.pig.load.HBaseLoader' with arguments XXX”的原因之一
你也许早就知道了：Pig可以加载HBase数据，从而更方便地进行数据处理。但是在使用HBase的loader的时候，可能会遇到这样那样的问题，我这里就遇到了一例，给大家分析一下原因。
使用 org.apache.pig.backend.hadoop.hbase.HBaseStorage() 可以加载HBase数据，例如：

A = LOAD 'hbase://table_name' USING org.apache.pig.backend.hadoop.hbase.HBaseStorage('column_family_name:qualifier_name', '-loadKey true -limit 100') AS (col1: chararray, col2:chararray);

其中，table_name 是你要加载数据的HBase表名，column_family_name:qualifier_name 是表的column family:qualifier（当然，可以有多个column family:qualifier，以空格隔开即可），-loadKey true -limit 100 是加载数据时指定的参数，支持的参数如下：

-loadKey=(true|false)  Load the row key as the first column

-gt=minKeyVal

-lt=maxKeyVal 

-gte=minKeyVal

-lte=maxKeyVal

-limit=numRowsPerRegion max number of rows to retrieve per region

-delim=char delimiter to use when parsing column names (default is space or comma)

-ignoreWhitespace=(true|false) ignore spaces when parsing column names (default true)

-caching=numRows  number of rows to cache (faster scans, more memory).

-noWAL=(true|false) Sets the write ahead to false for faster loading.

    To be used with extreme caution, since this could result in data loss

    (see http://hbase.apache.org/book.html#perf.hbase.client.putwal).

ERROR 2999: Unexpected internal error. could not instantiate 'com.twitter.elephantbird.pig.load.HBaseLoader' with arguments XXX
Details at logfile: XXX

m_conf = HBaseConfiguration.create();

可见它调用了HBase代码中的一个类HBaseConfiguration的create方法。按上面的提示，它是找不到这个方法，于是我们再看看使用的HBase的 HBaseConfiguration.java 里的代码，找遍全文，都找不到create方法！那么，我们再看看更新一点的版本的HBase的相同文件中是否有这个方法呢？下载0.90.4版本的HBase，果然在 HBaseConfiguration.java 中找到了create方法：

  /**                                                                                                                                                       
   * Creates a Configuration with HBase resources                                                                                                       
   * @return a Configuration with HBase resources                                                                                                       
   */
  public static Configuration create() {
    Configuration conf = new Configuration();
    return addHbaseResources(conf);
  }

所以，问题就在这里了：Pig的HBase loader不能使用某些版本的HBase，升级HBase吧！
另外，就算HBase版本适用了，你也得让Pig知道HBase的参数配置（要不然怎么可以指定一个HBase表名就可以加载其数据了呢），具体你可以看这个链接的说明。

✔ JOIN的优化
如果你对N个关系(relation)的某些字段进行JOIN，也就是所谓的“多路的”(multi-way)JOIN——我不知道用中文这样描述是否正确——在这种情况下，请遵循这样的原则来写JOIN语句：
JOIN用到的key所对应的记录最多的那个关系(relation)应该被排在最后。例如：

D = JOIN A BY col1, B BY col1, C BY col1;

在这里，假设C这个relation就是上面所说的那种情况，所以把它排在最后。
文章来源：http://www.codelast.com/
为什么要遵循这样的原则？这是由Pig处理JOIN的方式来决定的。在JOIN的n个关系中，前面的n-1个关系的处理结果会被cache在内存中，然后才会轮到第n个关系，因此，把最占内存的放在最后，有时候是能起到优化作用的。

✔ 错误“Backend error : org.apache.pig.data.BinSedesTuple cannot be cast to org.apache.pig.data.DataBag”的原因
如果你正在使用Pig 0.8，那么要注意了：出现这个错误，可能是Pig的bug导致的，详见这个链接。
说得简单点就是：此bug会导致无法解引用一个tuple中的bag。通常我们取一个tuple中的bag，是为了FLATTEN它，将记录展开，但是此bug使得你根本连tuple中的bag都输出不了。
此bug并不会影响你的Pig脚本语法解析，也就是说，你的Pig脚本只要写对了，就能运行起来，但是它执行到后面会报错。

✔ 如何加载LZO压缩的纯文本数据
如果你的数据是纯文本经由LZO压缩而成，那么你可以用elephant-bird的 com.twitter.elephantbird.pig.store.LzoPigStorage 来加载它：

A = LOAD '/user/codelast/text-lzo-file' USING com.twitter.elephantbird.pig.store.LzoPigStorage() AS (col1: chararray, col2: int);

注意，这里没有写REGISTER jar包的命令，你需要自己补上。

✔ 如何用Pig设置map端的并行度(map数)
从这个链接中的第(9)条，我们知道，无法通过PARALLEL来设置Pig job map端的并行度，但是，有没有什么办法可以间接实现这一点呢？
在Java编写的MapReduce程序中，你可以像这个链接中的第(25)点所说的一样，通过FileInputFormat.setMinInputSplitSize()来间接更改map的数量，其实它就与设置 mapred.min.split.size 参数值的效果是一样的。
在Pig中，我们是可以通过set命令来设置job参数的，所以，我们如果在Pig脚本的开头写上：

set mapred.min.split.size 2147483648;

将使得对map来说，小于2G的文件将被作为一个split输入，从而一个小于2G的文件将只有一个map。假设我们的Pig job是一个纯map的job，那么，map数的减少将使得输出文件的数量减少，在某些情况下，这种功能还是很有用的。
注意：上面的命令中，set的参数的单位是字节，所以2G=2*1024*1024*1024=2147483648。

✔ Pig调用现存的静态Java方法
不是每个人都会开发UDF，或者每个人都愿意去写一个UDF来完成一件极其简单的操作的，例如，把一个编码过的URL解码，如果我只想临时用一下这个功能，那么我还要去写一个UDF，累不累啊？
我们知道，java.net.URLDecoder.decode 这个静态方法已经实现了URL解码功能：

static String decode(String s, String enc) 

使用指定的编码机制对 application/x-www-form-urlencoded 字符串解码

那么，如何在Pig中使用这个现成的静态方法呢？为了展示这个使用过程，我造了一个数据文件：

[root@localhost ~]$ cat a.txt 
http://zh.wikipedia.org/zh/%E6%90%9C%E7%B4%A2%E5%BC%95%E6%93%8E

就一行，这个URL解码之后应该是：
http://zh.wikipedia.org/wiki/搜索引擎
因为里面含中文，所以被编码了。
处理此文件的Pig脚本url.pig如下：

DEFINE DecodeURL InvokeForString('java.net.URLDecoder.decode', 'String String');
A = LOAD 'a.txt' AS (url: chararray);
B = FOREACH A GENERATE DecodeURL(url, 'UTF-8');
STORE B INTO 'url';

文章来源：http://www.codelast.com/
用 pig -x local url.pig 执行这个脚本，完成后我们查看输出目录下的 part-m-00000 文件内容，可见它确实被解码成了正确的字符串。
这样，我们就利用了现存的静态Java方法来偷了个懒，很方便。
需要注意的是：只能调用静态方法，并且此调用比同样的UDF实现速度要慢，这是因为调用器没有使用Accumulator或Algebraic接口。根据这位兄台的测试，百万条的记录规模下，调用Java静态方法比UDF大约要慢一倍。至于这样的代价能不能接受，就看你自己的判断了。
文章来源：http://www.codelast.com/
✔ REGISTER UDF的前后覆盖关系
在Pig脚本中使用多个UDF时，我们通常会使用通配符的方式来引入：

REGISTER /user/hadoop/*.jar

假设你有另一个包含UDF的jar：/user/codelast/my-pig-lib.jar
并且这个jar的旧版本已经包含在 /user/hadoop/*.jar 中，那么，当你使用下面的方式来引入jar的时候：

REGISTER /user/hadoop/*.jar
REGISTER /user/codelast/my-pig-lib.jar

my-pig-lib.jar这个jar，使用的是新版还是旧版呢？这是不一定的，并不是说以最后REGISTER的为准，而是取决于CLASSPATH中jar的顺序。所以，如果你想保证能用到新版的UDF，为了保险你应该不要把旧版的UDF用 /user/hadoop/*.jar 的方式引入，而应该把 my-pig-lib.jar 从 /user/hadoop/*.jar 中剔除出去再REGISTER。

文章来源：https://www.codelast.com/
➤➤ 版权声明 ➤➤ 
转载需注明出处：codelast.com 
感谢关注我的微信公众号（微信扫一扫）：

以及我的微信视频号：