$ brew install pig

grunt> records = LOAD 'input/ncdc/micro-tab/sample.txt' AS (year:int, temperature:int, quality:int);
grunt> DESCRIBE records;
records: {year:int,temperature:int,quality:int}

注：不指定类型的话，默认是bytearray
    DESCRIBE 用来查看模式 (Schema)

$ cat max_temp.macro
DEFINE max_by_group(X, group_key, max_field) RETURNS Y {
A = GROUP $X by $group_key;
$Y = FOREACH A GENERATE group, MAX($X.$max_field);
};

grunt> IMPORT './max_temp.macro';

grunt> records = LOAD '/input/ncdc/micro-tab/sample.txt' AS (year:chararray, temperature:int, quality:int);
grunt> max_temp = max_by_group(records, year, temperature);
grunt> DUMP max_temp;

$ cat com/hadoopbook/pig/IsGoodQuality.java
package com.hadoopbook.pig;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.pig.FilterFunc;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.data.DataType;
import org.apache.pig.data.Tuple;
import org.apache.pig.impl.logicalLayer.FrontendException;

public class IsGoodQuality extends FilterFunc {

  @Override
  public Boolean exec(Tuple tuple) throws IOException {
    if (tuple == null || tuple.size() == 0) {
      return false;
    }
    try {
      Object object = tuple.get(0);
      if (object == null) {
        return false;
      }
      int i = (Integer) object;
      return i == 0 || i == 1 || i == 4 || i == 5 || i == 9;
    } catch (ExecException e) {
      throw new IOException(e);
    }
  }
}

$ javac -cp $(brew --prefix pig)/libexec/pig-0.16.0-core-h2.jar:$(hadoop classpath) com/hadoopbook/pig/IsGoodQuality.java
$ jar -cf com/hadoopbook/pig/IsGoodQuality.jar com/hadoopbook/pig/IsGoodQuality.class

注：我通过brew安装的pig所以使用了$(brew --prefix pig)；可以手动指定pig.jar的具体路径。

$ pig
grunt> REGISTER ./com/hadoopbook/pig/IsGoodQuality.jar;
grunt> DEFINE isGood com.hadoopbook.pig.IsGoodQuality();
grunt> records = LOAD '/input/ncdc/micro-tab/sample.txt' AS (year:chararray, temperature:int, quality:int);
grunt> filtered_records = FILTER records BY temperature != 9999 AND isGood(quality);
grunt> DUMP filtered_records;

$ cat max_temp_filter_udf.pig
--max_temp_filter_udf.pig
REGISTER com/hadoopbook/pig/IsGoodQuality.jar;
DEFINE isGood com.hadoopbook.pig.IsGoodQuality();
records = LOAD '/input/ncdc/micro-tab/sample.txt' AS (year:chararray, temperature:int, quality:int);
filtered_records = FILTER records BY temperature != 9999 AND isGood(quality);
grouped_records = GROUP filtered_records BY year;
max_temp = FOREACH grouped_records GENERATE group, MAX(filtered_records.temperature);
DUMP max_temp;
STORE max_temp INTO 'max_temp_output' USING PigStorage(':');

grunt> STORE max_temp INTO 'max_temp_output' USING PigStorage(':');
grunt> cat max_temp_output

grunt> max_temp = FOREACH grouped_records GENERATE group, MAX(filtered_records.temperature);

grunt> records = LOAD '/input/ncdc/micro-tab/sample.txt' AS (year:chararray, temperature:int, quality:int);
grunt> second_field = STREAM records THROUGH `cut -f 2`;
grunt> DUMP second_field;

$ cat is_good_quality.py
#!/usr/bin/env python
# encoding: utf-8
import re
import sys

for line in sys.stdin:
    (year, temp, q) = line.strip().split()
    if temp != '9999' and re.match('[01459]', q):
        print('{}\t{}'.format(year, temp))

$ cat max_temp_filter_stream.pig
--max_temp_filter_stream.pig

DEFINE is_good_quality `is_good_quality.py` SHIP ('is_good_quality.py');
records = LOAD '/input/ncdc/micro-tab/sample.txt' AS (year:chararray, temperature:int, quality:int);
filtered_records = STREAM records THROUGH is_good_quality AS (year:chararray, temperature:int);
grouped_records = GROUP filtered_records BY year;
max_temp = FOREACH grouped_records GENERATE group, MAX(filtered_records.temperature);
DUMP max_temp;
STORE max_temp INTO 'max_temp_stream';

$ hdfs dfs -put max_temp_filter_stream.pig
$ pig -f max_temp_filter_stream.pig
$ hdfs dfs -cat max_temp_stream/*
1949    111
1950    22

grunt> C = JOIN A BY $0, B BY $1;

注：等式连接，即：连接A.$0 == B.$1 的行，结果中的字段由所有输入关系的所有字段组成。

grunt> C = JOIN A BY $0, B BY $1 USING "replicated";

注：第一个关系必须是大的关系，后面则是一个或多个相对较小的关系。(能够全部放入内存)

grunt> grouped_records = GROUP records BY year PARALLEL 30;

grunt> set default_parallel 30;

$ cat max_temp_param.pig
records = LOAD '$input' AS (year:chararray, temperature:int, quality:int);
filtered_records = FILTER records BY temperature != 9999 AND (quality == 0 OR quality == 1 OR quality == 4 OR quality == 5 OR quality == 9);
grouped_records = GROUP filtered_records BY year;
max_temp = FOREACH grouped_records GENERATE group, MAX(filtered_records.temperature);
STORE max_temp INTO '$output';

$ pig -param input=/input/ncdc/micro-tab/sample.txt -param output=/tmp/out max_temp_param.pig

$ cat max_temp_param.param
# Input file
input=/input/ncdc/micro-tab/sample.txt
# Output file
output=/tmp/out

$ pig -param_file max_temp_param.param max_temp_param.pig