Apache Pig-用户定义函数

Apache Pig是一个可扩展和高级的平台，用于创建大规模数据处理应用程序，它可以运行在Hadoop生态系统内。Pig Latin是Pig的脚本语言，是一种高级的查询语言，使开发者可以更轻松的进行复杂数据分析。

Pig中自带了一些内置函数，如COUNT、SUM、AVG等，同时也支持自定义函数。用户定义函数（User Defined Function，UDF）是一种强大的工具，使Pig可以处理大量的用户特定需求。

Pig UDF种类

Pig定义了三种类型的UDF，它们在使用方式和定义语法上有所不同：

• Scalar Functions
• Filter Functions
• Load/Store Functions

Scalar Functions

Scalar Functions是一种接受零到多个输入和返回一个输出的函数。这些函数用于在计算中处理一个数据元素。Scalar Functions中的UDF函数可以枚举、计算和转换单个元素。

定义Scalar Functions

定义Scalar函数需要实现EvalFunc类和getArgToFuncMapping方法

public class MyScalarFunction extends EvalFunc<String> {

   public String exec(Tuple input) throws IOException {
      if (input == null || input.size() == 0)
         return null;
      try{
         String str = (String)input.get(0);
         String reversed = new StringBuffer(str).reverse().toString();
         return reversed;
      }catch(Exception e){
         throw WrappedIOException.wrap("Caught exception processing input row ", e);
      }
   }

   @Override
   public List<String> getArgToFuncMapping() {
      List<String> fields = new ArrayList<String>();
      fields.add("field1");
      fields.add("field2");
      return fields;
   }
}

参数说明：

• Tuple类型的参数input：输入元组
• 返回String类型：输出结果

Filter Functions

Filter Functions是一种接受一行输入和返回一个布尔输出的函数。它们主要用于过滤输入数据并确定哪些部分需要排除。

定义Filter Functions

和Scalar Functions相似，Filter Functions方法需要实现Func或org.apache.pig.FilterFunc，以下是使用FilterFunc的示例：

public class FileNameFilter extends FilterFunc {
 
   public Boolean exec(Tuple arg0) throws IOException {
      if (arg0 == null || arg0.size() == 0)
         return false;

      try {
         String fileName = (String)arg0.get(0);
         String extension = fileName.substring(fileName.lastIndexOf('.') + 1);
         return (extension.equals("txt")) ? true : false;
      } catch (Exception e) {
         throw WrappedIOException.wrap("Caught exception processing input row ", e);
      }
   }
}

Load/Store Functions

Load and Store functions是用于从外部数据源（如数据库、文件等）加载数据和将数据保存回外部数据源的函数。自定义Load和Store函数可以用来增加Pig的输出和输入源。

定义Load Functions

定义Load Functions需要实现LoadFunc类和getInputFormat方法。

public class MyLoadFunc extends LoadFunc {

   private RecordReader reader;

   public InputFormat getInputFormat() throws IOException {
      return new TextInputFormat();
   }

   public Tuple getNext() throws IOException {
      TupleFactory factory = TupleFactory.getInstance();
      Tuple tuple = factory.newTuple();

      try {
         boolean hasNext = reader.nextKeyValue();
         if (!hasNext) {
            return null;
         }
         Text value = (Text)reader.getCurrentValue();
         if (value == null || value.toString() == null) {
            return null;
         }
         String fields[] = value.toString().split(" ");
         for (int i = 0; i < fields.length; i++) {
            tuple.append(fields[i]);
         }
         return tuple;
      } catch (InterruptedException e) {
         throw new IOException(e);
      }
   }

   public void prepareToRead(RecordReader reader, PigSplit split)  throws IOException {
      this.reader = reader;
   }

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

参数说明：

• InputFormat类型：getInputFormat()方法返回输入格式
• 参数1 (RecordReader reader): RecordReader对象读取输入
• 参数2 (PigSplit split): PigSplit对象用于用于协调数据移动和处理作业
• 参数3 String location: 输入数据的位置
• 参数4 Job job: Job对象

定义Store Functions

定义 Store Functions 需要实现 StoreFunc 类和下面的 PutNext 方法。

public void putNext(Tuple tupleToWrite) throws IOException {
   try {
      StringOutputCollector.collect(mMos, tupleToWrite);
   } catch (InterruptedException e) {
      throw new IOException(e);
   }
}

使用UDF

使用UDF有两种基本方式：

1. 脚本中引用UDF
2. 在Java代码中执行UDF

脚本中引用UDF

在Pig Latin脚本中，用户定义的函数可以像内建函数一样调用。

REGISTER myudf.jar;
DEFINE myudf MyScalarFunction();
A = LOAD 'input.txt' AS (field1:chararray, field2:chararray);
B = FOREACH A GENERATE field1, myudf(field2);

参数说明：

• 参数1 myudf.jar: 定义储存UDF的位置
• 参数2 MyScalarFunction(): 指定使用的用户自定义函数

Java代码中调用UDF

我们在Java代码中通过PigServer.registerJar()注册自己写的jar包，以便在Pig脚本中使用自定义的UDF

Properties props = new Properties();
props.setProperty("fs.default.name", "hdfs://192.168.2.100:9000");
props.setProperty("mapred.job.tracker", "192.168.2.100:8021");
PigServer pigServer = new PigServer(ExecType.MAPREDUCE, props);
pigServer.registerJar("/home/hadoop/myudf.jar");

pigServer.registerFunction("myudf", "MyScalarFunction");

pigServer.registerQuery("A = LOAD 'input.txt' as (field1:chararray, field2:chararray);");
pigServer.registerQuery("B = FOREACH A GENERATE field1, myudf(field2);");

Iterator it = pigServer.openIterator("B");

while(it.hasNext()){
   System.out.println(it.next().toString());
}

以上代码实现的功能和上面脚本中的代码功能是一样的。

总结

在Pig的大数据处理中，UDF既可以让你开发自己的函数完成特定的需求分析，还可以使Pig更加灵活地实现各项功能。在大规模数据分析和处理工作时，UDF将使得工作变得更高效、更方便，更符合你的需求。