【大数据离线开发】9 Pig:数据分析引擎

这篇具有很好参考价值的文章主要介绍了【大数据离线开发】9 Pig:数据分析引擎。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

Pig:数据分析引擎

9.1 什么是Pig

9.1.1 简介

Pig是一个基于Apache Hadoop的大规模数据分析平台,它提供的SQL-LIKE语言叫PigLatin,该语言的编译器会把类SQL的数据分析请求转换为一系列经过优化处理的MapReduce运算。Pig为复杂的海量数据并行计算提供了一个简单的操作和编程接口,使用者可以透过Python或者JavaScript编写Java,之后再重新转写。

Pig的特点

  • Pig是一个用来处理大规模数据集的平台,由Yahoo!贡献给Apache

  • Pig可以简化MapReduce任务的开发

  • Pig可以看做hadoop的客户端软件,可以连接到hadoop集群迕行数据分析工作

  • Pig方便不熟悉java的用户,使用一种较为简便的类似二SQL的面向数据流的语言pig Latin迕行数据处理

  • PigLatin可以迕行排序、过滤、求和、分组、关联等常用操作,迓可以自定义凼数,返是一种面向数据分析处理的轻量级脚本语言

  • Pig可以看做是PigLatin到MapReduce的映射器

  • Pig可以自动对集群进行分配和回收,自动地对MapReduce程序进行优化

9.1.2 与 Hive 的对比

Pig与Hive作为一种高级数据语言,均运行于HDFS之上,是hadoop上层的衍生架构,用于简化hadoop任务,并对MapReduce进行一个更高层次的封装。Pig与Hive的区别如下:

  • Pig是一种面向过程的数据流语言;Hive是一种数据仓库语言,并提供了完整的sql查询功能。
  • Pig更轻量级,执行效率更快,适用于实时分析;Hive适用于离线数据分析。
  • Hive查询语言为Hql,支持分区;Pig查询语言为Pig Latin,不支持分区。
  • Hive支持JDBC/ODBC;Pig不支持JDBC/ODBC。
  • Pig适用于半结构化数据(如:日志文件);Hive适用于结构化数据。

9.2 Pig的体系结构和数据模型

为了使用Pig执行特定任务,程序员需要使用Pig Latin语言编写Pig脚本,并使用任何执行机制(Grunt Shell,UDF,嵌入式)执行它们。执行后,这些脚本将经历Pig框架应用的一系列转换,以产生所需的输出。

在内部,Apache Pig将这些脚本转换为一系列MapReduce作业,因此使程序员的工作变得容易。Apache Pig的体系结构如下所示。

【大数据离线开发】9 Pig:数据分析引擎

数据模型

【大数据离线开发】9 Pig:数据分析引擎

9.3 Pig的安装和工作模式

9.3.1 Pig安装

首先将Pig的包上传到我的Linux目录当中,然后对文件进行解压安装

【大数据离线开发】9 Pig:数据分析引擎

解压命令
tar -zxvf pig-0.17.0.tar.gz -C ~/training/

设置环境变量 vi ~/.bash_profile
PIG_HOME=/root/training/pig-0.17.0
export PIG_HOME
PATH=$PIG_HOME/bin:$PATH
export PATH
生效环境变量 source ~/.bash_profile

Pig的常用命令

  • ls、cd、cat、mkdir、pwd
  • copyFromLocal、copyToLocal
  • sh
  • register、define
9.3.2 Pig工作模式

本地模式:操作Linux的文件

启动: pig -x local

日志:Connecting to hadoop file system at: file:///

【大数据离线开发】9 Pig:数据分析引擎

集群模式:链接到HDFS

设置环境变量PIG_CLASSPATH指向Hadoop配置文件所在的目录

注意:通过Pig操作HDFS的速度要快

设置环境变量 vi ~/.bash_profile
PIG_CLASSPATH=/root/training/hadoop-2.7.3/etc/hadoop
export PIG_CLASSPATH
生效环境变量 source ~/.bash_profile

启动: pig

日志: Connecting to hadoop file system at: hdfs://bigdata111:9000

【大数据离线开发】9 Pig:数据分析引擎

9.4 Pig的内置函数

计算函数

函数名称 函数作用
avg 计算包中项的平均值
concat 把两个字节数组或者字符数组连接成一个
count 计算包中非空值的个数
count_star 计算包中项的个数,包括空值
diff 计算两个包的差
max 计算包中项的最大值
min 计算包中项的最小值
size 计算一个类型的大小,数值型的大小为1; 对于字符数组,返回字符的个数; 对于字节数组,返回字节的个数; 对于元组,包,映射,返回其中项的个数。
sum 计算一个包中项的值的总和
TOKENIZE 对一个字符数组进行标记解析,并把结果词放入一个包

过滤函数

函数名称 函数作用
isempty 判断一个包或映射是否为空

加载存储函数

函数名称 函数作用
PigStorage 用字段分隔文本格式加载或存储关系,这是默认的存储函数
BinStorage 从二进制文件加载一个关系或者把关系存储到二进制文件
BinaryStorage 从二进制文件加载只是包含一个类型为bytearray的字段的元组到关系,或以这种格式存储一个关系
TextLoader 从纯文本格式加载一个关系
PigDump 用元组的tostring()形式存储关系

9.5 使用PigLatin语句分析数据

注意:类似Spark RDD的算子(方法、函数)

​ Spark的算子有两种类型:

  1. Transformation:不会触发计算
  2. Action:会触发计算

注意:启动Yarn的history server

  1. 首先需要使用Hadoop的HistoryServer,前提要启动hadoop环境

    • 命令:mr-jobhistory-daemon.sh start historyserver

    • 网页:http://192.168.157.111:19888/jobhistory(包含所有的HDFS操作)

  1. 常用的PigLatin语句:有的会触发计算,有的不会

    名称 作用 Spark对应语句
    load 加载数据到bag(表)
    foreach 相当于循环,对bag每一条数据tuple进行处理 RDD.foeach
    filter 过滤:相当于where RDD.filter
    group by 分组 也有
    join 连接操作 也有
    generate 提取列
    union/intersect 集合运算
    dump 输出:直接打印的屏幕上
    store 输出: 输出到文件

  2. 举例:数据(员工表、部门表)

    7654,MARTIN,SALESMAN,7698,1981/9/28,1250,1400,30

(1) 加载员工数据到表
	emp = load '/scott/emp.csv';
	查询表的结构
	describe emp;  ---> Schema for emp unknown.
	
(2) 加载员工数据到表,指定每个tuple的结构和类型
	emp = load '/scott/emp.csv' as(empno,ename,job,mgr,hiredate,sal,comm,deptno);
	默认的数据类型:bytearray
	默认分隔符:制表符
	
    emp = load '/scott/emp.csv' as(empno:int,ename:chararray,job:chararray,mgr:int,hiredate:chararray,sal:int,comm:int,deptno:int);
	
	emp = load '/scott/emp.csv' using PigStorage(',') as(empno:int,ename:chararray,job:chararray,mgr:int,hiredate:chararray,sal:int,comm:int,deptno:int);
	
	创建一个部门表
	dept = load '/scott/dept.csv' using PigStorage(',') as(deptno:int,dname:chararray,loc:chararray);
	
(3) 查询员工信息:员工号 姓名  薪水
	SQL: select empno,ename,sal from emp;
	PL:  emp3 = foreach emp generate empno,ename,sal;
	
(4) 查询员工信息:按照月薪排序
	SQL: select * from emp order by sal;
	PL:  emp4 = order emp by sal;
	
(5) 分组:求每个部门的工资的最大值
	SQL:  select deptno,max(sal) from emp group by deptno;
	PL:   第一步:分组
		  emp51 = group emp by deptno;
		  表结构:
		  emp51: {group: int,
                  emp: {(empno: int,ename: chararray,job: chararray,mgr: int,hiredate: chararray,sal: int,comm: int,deptno: int)}}
		  
		  数据:
		  (10,{(7934,MILLER,CLERK,7782,1982/1/23,1300,,10),
			   (7839,KING,PRESIDENT,,1981/11/17,5000,,10),
			   (7782,CLARK,MANAGER,7839,1981/6/9,2450,,10)})
			   
		  (20,{(7876,ADAMS,CLERK,7788,1987/5/23,1100,,20),
			   (7788,SCOTT,ANALYST,7566,1987/4/19,3000,,20),
			   (7369,SMITH,CLERK,7902,1980/12/17,800,,20),
			   (7566,JONES,MANAGER,7839,1981/4/2,2975,,20),
			   (7902,FORD,ANALYST,7566,1981/12/3,3000,,20)})
			   
		  (30,{(7844,TURNER,SALESMAN,7698,1981/9/8,1500,0,30),
			   (7499,ALLEN,SALESMAN,7698,1981/2/20,1600,300,30),
			   (7698,BLAKE,MANAGER,7839,1981/5/1,2850,,30),
			   (7654,MARTIN,SALESMAN,7698,1981/9/28,1250,1400,30),
			   (7521,WARD,SALESMAN,7698,1981/2/22,1250,500,30),
			   (7900,JAMES,CLERK,7698,1981/12/3,950,,30)})
		  
		  第二步:求每个部门的工资最大值
			  emp52 = foreach emp51 generate group,MAX(emp.sal)
	
	(6) 查询10号部门的员工
		SQL:  select * from emp where deptno=10;
		PL:   emp6 = filter emp by deptno==10;  注意:两个等号
		
	(7) 多表查询
		查询员工信息: 员工姓名 部门名称
		SQL:  select e.ename,d.dname from emp e,dept d where e.deptno=d.deptno;
		PL:   emp71 = join dept by deptno,emp by deptno;
			  emp72 = foreach emp71 generate dept::dname,emp::ename;
			  
	(8) 集合运算:关系型数据库Oracle:参与集合运算的各个集合必须列数相同且类型一致
		 1020号部门的员工
		 SQL: select * from emp where deptno=10
			  union
			  select * from emp where deptno=20;
			  
		PL: emp10 = filter emp by deptno==10;
			emp20 = filter emp by deptno==20;
			emp10_20 = union emp10,emp20;
			
	(9) 使用PL实现WordCount
		① 加载数据 
		mydata = load '/data/data.txt' as (line:chararray);

		② 将字符串分割成单词 
		words = foreach mydata generate flatten(TOKENIZE(line)) as word;

		③ 对单词进行分组 
		grpd = group words by word; 

		④ 统计每组中单词数量 
		cntd = foreach grpd generate group,COUNT(words); 

		⑤ 打印结果 
		dump cntd;	
        
注意:PigLatin中,bag具有依赖关系
	  Spark中,RDD也具有依赖关系(宽依赖、窄依赖)

【大数据离线开发】9 Pig:数据分析引擎

9.6 Pig的自定义函数

Pig的自定义函数有三种:

  1. 自定义过滤函数:相当于where条件

  2. 自定义运算函数:

  3. 自定义加载函数:使用load语句加载数据,生成一个bag

    默认:load语句加载数据,一行解析成一个Tuple

    需要MR的jar包

9.6.1 自定义过滤和运算函数

依赖的jar包

1、/root/training/pig-0.17.0/pig-0.17.0-core-h2.jar

2、/root/training/pig-0.17.0/lib

3、/root/training/pig-0.17.0/lib/h2

4、$HADOOP_HOME/share/hadoop/common

5、$HADOOP_HOME/share/hadoop/common/lib

自定义过滤函数IsSalaryTooHigh

//实现自定义的过滤函数:实现:查询过滤薪水大于2000的员工
public class IsSalaryTooHigh extends FilterFunc{

	@Override
	public Boolean exec(Tuple tuple) throws IOException {
		/*参数tuple,调用的时候 传递的参数
		 * 
		 * 在PigLatin调用:
		 * myresult1 = filter emp by demo.pig.IsSalaryTooHigh(sal)
		 */
		//取出薪水
		int sal = (int) tuple.get(0);
		return sal>2000?true:false;
	}

}

自定义过滤函数CheckSalaryGrade

//根据员工的薪水,判断级别
//泛型:经过运算后,结果的类型
public class CheckSalaryGrade extends EvalFunc<String>{

	@Override
	public String exec(Tuple tuple) throws IOException {
		// 调用: myresult2 = foreach emp generate ename,sal,demo.pig.CheckSalaryGrade(sal);
		
		int sal = (int)tuple.get(0);
		if(sal<1000) return "Grade A";
		else if(sal>=1000 && sal<3000) return "Grade B";
		else return "Grade C";
	}

}

然后将部署的java代码打包成jar包,上传到Linux服务器上,进行注册执行

  • 注册jar包 register /root/temp/pig.jar

  • 为自定义函数起别名(可选):define myfilter demo.pig.MyFilterFunction;

  • 用法1:result1 = filter emp by demo.pig.IsSalaryTooHigh(sal)

  • 用法2:result2 = foreach emp generate ename,demo.pig.CheckSalaryGrade(sal);

【大数据离线开发】9 Pig:数据分析引擎

【大数据离线开发】9 Pig:数据分析引擎

9.6.2 自定义加载函数
  • 默认情况下,load语句加载数据,一行解析成一个Tuple

    比如:员工信息

【大数据离线开发】9 Pig:数据分析引擎

  • 特殊情况:单词统计的时候。这时候:希望每个单词能被解析成一个Tuple,从而便于处理

【大数据离线开发】9 Pig:数据分析引擎

  • 需要MapReduce的 jar 包(参考之前下载MapReduce的jar选择)

【大数据离线开发】9 Pig:数据分析引擎

自定义加载函数MyLoadFunc

/*
 * 自定义的加载函数:
 * 实现:WordCount钱加载数据,每一个单词作为一个Tuple
 */
public class MyLoadFunc extends LoadFunc{

	//定义一个变量保存输入流
	private RecordReader reader ;
	
	@Override
	public InputFormat getInputFormat() throws IOException {
		// 输入数据的格式:字符串
		return new TextInputFormat();
	}

	@Override
	public Tuple getNext() throws IOException {
		// 从输入流中读取一行,如何解析生成返回的tuple
		//数据: I love Beijing
		Tuple result = null;
		try{
			//判断是否读入了数据
			if(!this.reader.nextKeyValue()){
				//没有数据
				return result; //----> 是null值
			}
			
			//数据数据: I love Beijing
			String data = this.reader.getCurrentValue().toString();
			
			//生成返回的结果:Tuple
			result = TupleFactory.getInstance().newTuple();
			
			//分词
			String[] words = data.split(" ");
			
			//每一个单词单独生成一个tuple(s),再把这些tuple放入一个bag中。
			//在把这个bag放入result中
			//创建一个表
			DataBag bag = BagFactory.getInstance().newDefaultBag();
			for(String w:words){
				//为每个单词生成tuple
				Tuple aTuple = TupleFactory.getInstance().newTuple();
				aTuple.append(w); //将单词放入tuple
				
				//再把这些tuple放入一个bag中
				bag.add(aTuple);
			}
			
			//在把这个bag放入result中
			result.append(bag);
			
		}catch(Exception ex){
			ex.printStackTrace();
		}
		
		return result;
	}

	@Override
	public void prepareToRead(RecordReader reader, PigSplit arg1) throws IOException {
		// 使用该方法,初始化输入流
        // RecordReader reader: 代表HDFS的输入流
		this.reader = reader;
	}

	@Override
	public void setLocation(String path, Job job) throws IOException {
		// 从HDFS输入的路径(目录)
		FileInputFormat.setInputPaths(job, new Path(path));
	}

}

然后将部署的java代码打包成jar包,上传到Linux服务器上,进行注册执行

注册jar包: register /root/temp/p1.jar

myresult3 = load ‘/input/data.txt’ using demo.pig.MyLoadFunc();

定义别名(可选):define myload demo.pig.MyLoadFunc;

【大数据离线开发】9 Pig:数据分析引擎文章来源地址https://www.toymoban.com/news/detail-436130.html

到了这里,关于【大数据离线开发】9 Pig:数据分析引擎的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包