Apache Pig
目录
- 一、配置说明
- 1.本地模式
- 2.集群模式
- 二、pig的数据模型
- 三、pig的数据类型
- 四、惰性执行
- 五、pig的基本语法
- 5.1语法说明
- 5.2案例操作
- 六、pig的自定义函数
一、配置说明
1.本地模式
操作的是Linux系统文件
pig -x local
关键日志
当前处于root目录下
2.集群模式
连接的是HDFS
相较于本地模式需要新增一个环境变量，指向HDFS目录
PIG_CLASSPATH=$HADOOP_HOME/etc/hadoop
export PIG_CLASSPATH
启动
pig
关键日志
二、pig的数据模型
三、pig的数据类型
四、惰性执行
含义
Pig采用了惰性执行（lazy evaluation）的策略。惰性执行意味着Pig会尽可能推迟实际的数据处理操作，直到它认为必须执行这些操作为止。
惰性执行的原因
Pig之所以采用惰性执行，主要是为了优化性能和提高效率。通过推迟不必要的操作，Pig可以减少数据的读取、写入和传输次数，从而节省时间和资源。
触发操作
存储操作：当你使用STORE语句将关系的数据存储到文件系统或其他存储系统时，Pig会触发实际的数据处理操作。
DUMP操作：当你使用DUMP语句时，Pig会输出关系中的数据到控制台或指定的输出文件，这也会触发实际的操作。
需要中间结果的操作：在某些情况下，Pig可能需要中间结果来执行后续的操作（比如连接、分组等），这时它也会触发必要的数据处理。
显式调用执行：在某些Pig实现或上下文中，你可能可以显式地调用执行操作（比如通过API调用），但这通常不是Pig脚本的常规用法。
五、pig的基本语法
5.1语法说明
Pig Latin关系操作
Pig Latin的诊断操作
Pig Latin UDF语句
5.2案例操作
DUMP输出关系中的所有数据
dump emp;
Pig会将emp关系中的所有数据输出到控制台或指定的输出文件中。通常，这对于小数据集来说是有用的，但是对于大数据集来说，可能会产生大量的输出
DESC显示关系的模式，包括字段名、数据类型和是否允许为null
desc emp;
emp: {ename: chararray, deptno: int, sal: float}
这里，ename、deptno 和 sal 是emp关系中的字段名，chararray、int 和 float 是它们对应的数据类型。
加载到表
emp = LOAD '/scott/emp.csv' USING PigStorage(',') AS (empno: int, ename: chararray, job: chararray, mgr: int, hiredate: chararray, sal: int, comm: int, deptno: int);
LOAD 是正确的关键字，用于加载数据。
‘/scott/emp.csv’ 是文件路径，指定了要加载的CSV文件的位置。
USING PigStorage(‘,’) 指定了使用 PigStorage 函数，并以逗号（,）作为字段分隔符。注意，逗号和括号之间应该有空格。
AS (empno: int, ename: chararray, job: chararray, mgr: int, hiredate: chararray, sal: int, comm: int, deptno: int) 定义了关系 emp 中的字段名称和数据类型。每个字段名称和数据类型之间用冒号（:）分隔，并且整个 AS 子句用括号包围。
遍历
emp3 = foreach emp generate empno, ename, sal;
遍历emp关系中的每一行，并生成一个新的关系emp3，其中包含empno、ename和sal三个字段。
排序
排序操作可能会消耗大量的计算资源，特别是当关系非常大时
emp4 = order emp by deptno, sal;
emp4 关系中的元组会首先根据 deptno 字段进行排序，然后在每个相同的 deptno 值内，根据 sal 字段进行排序。
emp4 = order emp by sal DESC;
这些元组会根据 sal 字段的值进行排序。
默认情况下，排序是升序的（从小到大），此处增加DESC为降序
分组及组内操作
1.分组
emp51 = group emp by deptno;
这回创建一个新的关系 emp51，其中包含根据 deptno 字段分组的元组。每个组由一个唯一的 deptno 值和该值对应的所有元组组成。在Pig中，分组后的结果通常是一个包含两个字段的关系：一个用于分组的字段（在这个例子中是 deptno），另一个是一个包（bag），包含该组中的所有原始元组。
2.组内操作——求部门最高工资
emp52 = foreach emp51 generate group as deptno, MAX(emp.sal) as max_sal;
foreach语句用于遍历emp51中的每个元组（在这里，每个元组代表一个部门及其员工数据）。
generate语句用于生成新的字段。在这里，group字段被重命名为deptno，并且使用MAX函数计算包（bag）中emp.sal字段的最大值，结果命名为max_sal。
emp_max_sal：包含字段deptno和max_sal的新关系。每个deptno值对应一个max_sal值，表示该部门的最高薪资。
查询
emp6 = filter emp by deptno==10;
filter 语句用于根据一个或多个条件筛选关系中的元组。此语句用于从 emp 关系中筛选出 deptno 字段等于 10 的所有元组，并将结果存储在新的关系 emp6 中。
多表查询
1.表格连接
emp71 = join dept by deptno,emp by deptno
join语句用于根据一个或多个公共字段将两个或多个关系（relations）合并在一起。此语句根据deptno字段将dept关系和emp关系进行连接，并将结果存储在新的关系emp71中。
连接的字段只要数据类型相同，内容相同（逻辑上正确），名称允许不一样
如：emp71 = join dept by deptno,emp by departmentID 都是部门号
2.内容查询
emp72= foreach emp71 generate dept::dname ,emp::ename ;
从emp71关系中生成一个新的关系emp72，其中包含dept关系中的dname字段（部门名称）和emp关系中的ename字段（员工名称）。
dept::dname：这表示从dept关系中选择dname字段。在emp71关系中，由于dept和emp已经被连接，所以dept的字段可以通过dept::前缀来访问。
emp::ename：这表示从emp关系中选择ename字段。同样地，在emp71关系中，emp的字段可以通过emp::前缀来访问。
集合运算
-- 过滤出部门编号为10的员工
emp10 = filter emp by deptno == 10; -- 过滤出部门编号为20的员工
emp20 = filter emp by deptno == 20; -- 将两个过滤后的关系进行联合
emp10_20 = union emp10, emp20;
emp10 = filter emp by deptno 10;：这行代码从 emp 关系中过滤出部门编号为10的员工，并将结果存储在 emp10 关系中。 emp20 = filter emp by deptno 20;：这行代码从 emp 关系中过滤出部门编号为20的员工，并将结果存储在 emp20 关系中。
emp10_20 = union emp10, emp20;：这行代码将 emp10 和 emp20 两个关系进行联合，生成一个新的关系 emp10_20。这个新关系将包含所有部门编号为10和20的员工。
六、pig的自定义函数
一般而言，pig的函数分为4种类型
- 过滤函数
- 计算函数
- 加载函数
- 存储函数
过滤函数
import org.apache.pig.FilterFunc;
import org.apache.pig.data.Tuple;
import java.io.IOException; public class IsSalaryTooHigh extends FilterFunc { @Override public Boolean exec(Tuple tuple) throws IOException { if (tuple == null || tuple.size() == 0) { return false; // 如果tuple为空，返回false } // 获取当前员工的薪水，假设薪水字段为整数类型 Integer sal = (Integer) tuple.get(0); // 确保索引0对应的是薪水字段 // 判断薪水是否大于等于3000 return sal != null && sal >= 3000; }
}
继承自FilterFunc。FilterFunc是Apache Pig中的一个接口，用于定义过滤函数。
重写exec方法
Tuple：
一个Tuple通常对应于表中的一行数据。一张表，它包含了多个Tuple，每个Tuple都代表了表中的一行。
Tuple中的索引通常是从0开始的，每个索引都对应了一个字段。
Tuple中的每个字段则对应了该行中的一个数据项。
计算函数
import org.apache.pig.EvalFunc;
import org.apache.pig.data.Tuple;
import java.io.IOException; public class CheckSalaryGrade extends EvalFunc<String> { @Override public String exec(Tuple tuple) throws IOException { // 获取员工薪水，假设薪水在tuple的第一个位置 int sal = (Integer) tuple.get(0); if (sal <= 1000) { return "Grade A"; } else if (sal > 1000 && sal <= 3000) { return "Grade B"; } else { return "Grade C"; } }
}
继承自EvalFunc< String >。String表示经过运算返回的结果类型
重新exec方法
加载函数（不全）
public class MyLoadFunc extends LoadFunc { @Override public InputFormat getInputFormat() throws IOException { // 返回自定义InputFormat类，或者使用Hadoop的默认输入格式， // 返回相应的InputFormat类，比如TextInputFormat.class return null; // 这里只是占位，实际实现中应该返回具体的InputFormat实例 } @Override public Tuple getNext() throws IOException { // 从输入流中读取一行，并解析该行数据 if (reader != null) { // 假设您的数据是文本格式，每行代表一个Tuple // 这里只是示例，实际解析可能更复杂 LongWritable key = new LongWritable(); Text value = new Text(); boolean hasNext = reader.nextKeyValue(); if (hasNext) { key = reader.getCurrentKey(); value = reader.getCurrentValue(); // 根据实际情况解析value，并创建Tuple // 这里简单地将整行作为一个字符串返回 return TupleFactory.getInstance().newTuple(new Object[]{value.toString()}); } } return null; // 如果没有更多数据可读，返回null } } @Override public void prepareToRead(RecordReader reader, PigSplit split) throws IOException { // 这个方法在你开始读取数据之前被调用，通常用于初始化一些状态或资源 // PigSplit对象包含了关于数据分割的信息，这可以帮助你确定要读取哪些数据 } @Override public void setLocation(String location, Job job) throws IOException { // 这个方法用于设置输入数据的路径或位置 // 你可以在这里配置Job对象，以便它知道从哪里读取数据 }
}