Pig Latin 基础，Apache Pig 加载和存储数据(forty-seven day first)

梁池暝

2023-12-01

Pig Latin 基础

Pig Latin是用于使用Apache Pig分析Hadoop中数据的语言。在本章中，我们将讨论Pig Latin的基础知识，如Pig Latin语句，数据类型，通用运算符，关系运算符和Pig Latin UDF。

Pig Latin - 数据模型

如前面章节所讨论的，Pig的数据模型是完全嵌套的。Relation是Pig Latin数据模型的最外层结构。它是一个包其中:

包是元组的集合。
元组是有序的字段集。
字段是一段数据。

Pig Latin - 语句

在使用Pig Latin处理数据时，语句是基本结构。

这些语句使用关系（relation），它们包括表达式（expression）和模式（schema）。
每个语句以分号（;）结尾。
我们将使用Pig Latin提供的运算符通过语句执行各种操作。
除了LOAD和STORE，在执行所有其他操作时，Pig Latin语句采用关系作为输入，并产生另一个关系作为输出。
只要在Grunt shell中输入 Load 语句，就会执行语义检查。要查看模式的内容，需要使用 Dump 运算符。只有在执行 dump 操作后，才会执行将数据加载到文件系统的MapReduce作业。

例子

下面给出一个Pig Latin语句，它将数据加载到Apache Pig中。

grunt> Student_data = LOAD 'student_data.txt' USING PigStorage(',')as 
   ( id:int, firstname:chararray, lastname:chararray, phone:chararray, city:chararray );

Pig Latin - 数据类型

下面给出的表描述了Pig Latin数据类型。

序号	数据类型	说明＆示例
1	int	表示有符号的32位整数。示例:8
2	long	表示有符号的64位整数。示例:5L
3	float	表示有符号的32位浮点。示例:5.5F
4	double	表示64位浮点。示例:10.5
5	chararray	表示Unicode UTF-8格式的字符数组（字符串）。示例:‘w3cschool’
6	Bytearray	表示字节数组（blob）。
7	Boolean	表示布尔值。示例:true / false。
8	Datetime	表示日期时间。示例:1970-01-01T00:00:00.000 + 00:00
9	Biginteger	表示Java BigInteger。示例:60708090709
10	Bigdecimal	表示Java BigDecimal 示例:185.98376256272893883
复杂类型
11	Tuple	元组是有序的字段集。示例:（raja,30）
12	Bag	包是元组的集合。示例:{（raju,30）,（Mohhammad,45）}
13	Map	地图是一组键值对。示例:['name'＃'Raju','age'＃30]

Null值

所有上述数据类型的值可以为NULL。Apache Pig以与SQL类似的方式处理空值。null可以是未知值或不存在值，它用作可选值的占位符。这些空值可以自然出现或者可以是操作的结果。

Pig Latin - 算术运算符

下表描述了Pig Latin的算术运算符。假设a = 10和b = 20。

运算符	描述	示例
+	加 - 运算符的两侧的值相加	a+b将得出30
−	减 - 从运算符左边的数中减去右边的数	a-b将得出-10
*	乘 - 运算符两侧的值相乘	a*b将得出200
/	除 - 用运算符左边的数除右边的数	b / a将得出2
%	系数 - 用运算符右边的数除左边的数并返回余数	b％a将得出0
？ :	Bincond - 评估布尔运算符。它有三个操作数，如下所示。变量 x =（expression）？ value1（如果为true）: value2（如果为false）。	b =（a == 1）？ 20:30; 如果a = 1，则b的值为20。如果a！= 1，则b的值为30。
CASE WHEN THEN ELSE END	Case - case运算符等效于嵌套的bincond运算符。	CASE f2 ％ 2 WHEN 0 THEN 'even' WHEN 1 THEN 'odd' END

Pig Latin - 比较运算符

下表描述了Pig Latin的比较运算符。

运算符	描述	示例
==	等于 - 检查两个数的值是否相等；如果是，则条件变为true。	（a = b）不为true。
!=	不等于 - 检查两个数的值是否相等。如果值不相等，则条件为true。	（a！= b）为true。
>	大于 - 检查左边数的值是否大于右边数的值。如果是，则条件变为true。	（a> b）不为true。
<	小于 - 检查左边数的值是否小于右边数的值。如果是，则条件变为true。	（a<b）为true。
>=	大于或等于 - 检查左边数的值是否大于或等于右边数的值。如果是，则条件变为true。	（a>=b）不为true。
<=	小于或等于 - 检查左边数的值是否小于或等于右边数的值。如果是，则条件变为true。	（a<=b）为true。
matches	模式匹配 - 检查左侧的字符串是否与右侧的常量匹配。	f1 matches '.* tutorial.*'

Pig Latin - 类型结构运算符

下表描述了Pig Latin的类型结构运算符。

运算符	描述	示例
()	元组构造函数运算符 - 此运算符用于构建元组。	（Raju,30）
{}	包构造函数运算符 - 此运算符用于构造包。	{（Raju,30）,（Mohammad,45）}
[]	映射构造函数运算符 - 此运算符用于构造一个映射。	[name＃Raja,age＃30]

Pig Latin - 关系运算符

下表描述了Pig Latin的关系运算符。

运算符	描述
加载和存储
LOAD	将数据从文件系统（local/ HDFS）加载到关系中。
STORE	将数据从文件系统（local/ HDFS）存储到关系中。
过滤
FILTER	从关系中删除不需要的行。
DISTINCT	从关系中删除重复行。
FOREACH，GENERATE	基于数据列生成数据转换。
STREAM	使用外部程序转换关系。
分组和连接
JOIN	连接两个或多个关系。
COGROUP	将数据分组为两个或多个关系。
GROUP	在单个关系中对数据进行分组。
CROSS	创建两个或多个关系的向量积。
排序
ORDER	基于一个或多个字段（升序或降序）按排序排列关系。
LIMIT	从关系中获取有限数量的元组。
合并和拆分
UNION	将两个或多个关系合并为单个关系。
SPLIT	将单个关系拆分为两个或多个关系。
诊断运算符
DUMP	在控制台上打印关系的内容。
DESCRIBE	描述关系的模式。
EXPLAIN	查看逻辑，物理或MapReduce执行计划以计算关系。
ILLUSTRATE	查看一系列语句的分步执行。

Apache Pig 加载数据

一般来说，Apache Pig在Hadoop之上工作。它是一种分析工具，用于分析 Hadoop File System中存在的大型数据集。要使用Apache Pig分析数据，我们必须首先将数据加载到Apache Pig中。本章介绍如何从HDFS将数据加载到Apache Pig。

准备HDFS

在MapReduce模式下，Pig从HDFS读取（加载）数据并将结果存回HDFS。因此，让我们先从HDFS开始，在HDFS中创建以下示例数据。

学生ID	名字	姓氏	电话号码	城市
001	Rajiv	Reddy	9848022337	Hyderabad
002	siddarth	Battacharya	9848022338	Kolkata
003	Rajesh	Khanna	9848022339	Delhi
004	Preethi	Agarwal	9848022330	Pune
005	Trupthi	Mohanthy	9848022336	Bhuwaneshwar
006	Archana	Mishra	9848022335	Chennai

上述数据集包含六个学生的个人详细信息，如id，名字，姓氏，电话号码和城市。

步骤1:验证Hadoop

首先，使用Hadoop version命令验证安装，如下所示。

$ hadoop version

如果你的系统里有Hadoop，并且已设置PATH变量，那么你将获得以下输出 -

Hadoop 2.6.0 
Subversion https://git-wip-us.apache.org/repos/asf/hadoop.git -r 
e3496499ecb8d220fba99dc5ed4c99c8f9e33bb1 
Compiled by jenkins on 2014-11-13T21:10Z 
Compiled with protoc 2.5.0 
From source with checksum 18e43357c8f927c0695f1e9522859d6a 
This command was run using /home/Hadoop/hadoop/share/hadoop/common/hadoop
common-2.6.0.jar

步骤2:启动HDFS

浏览Hadoop的 sbin 目录，并启动 yarn 和Hadoop dfs（分布式文件系统），如下所示。

cd /$Hadoop_Home/sbin/ 
$ start-dfs.sh 
localhost: starting namenode, logging to /home/Hadoop/hadoop/logs/hadoopHadoop-namenode-localhost.localdomain.out 
localhost: starting datanode, logging to /home/Hadoop/hadoop/logs/hadoopHadoop-datanode-localhost.localdomain.out 
Starting secondary namenodes [0.0.0.0] 
starting secondarynamenode, logging to /home/Hadoop/hadoop/logs/hadoop-Hadoopsecondarynamenode-localhost.localdomain.out
 
$ start-yarn.sh 
starting yarn daemons 
starting resourcemanager, logging to /home/Hadoop/hadoop/logs/yarn-Hadoopresourcemanager-localhost.localdomain.out 
localhost: starting nodemanager, logging to /home/Hadoop/hadoop/logs/yarnHadoop-nodemanager-localhost.localdomain.out

步骤3:在HDFS中创建目录

在Hadoop DFS中，可以使用 mkdir 命令创建目录。在HDFS所需路径中创建一个名为 Pig_Data 的新目录，如下所示。

$cd /$Hadoop_Home/bin/ 
$ hdfs dfs -mkdir hdfs://localhost:9000/Pig_Data

步骤4:将数据放在HDFS中

Pig的输入文件包含单个行中的每个元组/记录。记录的实体由分隔符分隔（在我们的示例中，我们使用“，”）。在本地文件系统中，创建一个包含数据的输入文件 student_data.txt ，如下所示。

001,Rajiv,Reddy,9848022337,Hyderabad
002,siddarth,Battacharya,9848022338,Kolkata
003,Rajesh,Khanna,9848022339,Delhi
004,Preethi,Agarwal,9848022330,Pune
005,Trupthi,Mohanthy,9848022336,Bhuwaneshwar
006,Archana,Mishra,9848022335,Chennai.

现在，使用 put 命令将文件从本地文件系统移动到HDFS，如下所示。（你也可以使用 copyFromLocal 命令。）

$ cd $HADOOP_HOME/bin 
$ hdfs dfs -put /home/Hadoop/Pig/Pig_Data/student_data.txt dfs://localhost:9000/pig_data/

验证文件

使用 cat 命令验证文件是否已移入HDFS，如下所示。

$ cd $HADOOP_HOME/bin
$ hdfs dfs -cat hdfs://localhost:9000/pig_data/student_data.txt

输出

现在，可以看到文件的内容，如下所示。

15/10/01 12:16:55 WARN util.NativeCodeLoader: Unable to load native-hadoop
library for your platform... using builtin-java classes where applicable
  
001,Rajiv,Reddy,9848022337,Hyderabad
002,siddarth,Battacharya,9848022338,Kolkata
003,Rajesh,Khanna,9848022339,Delhi
004,Preethi,Agarwal,9848022330,Pune
005,Trupthi,Mohanthy,9848022336,Bhuwaneshwar
006,Archana,Mishra,9848022335,Chennai

Load运算符

你可以使用 Pig Latin 的 LOAD 运算符，从文件系统（HDFS / Local）将数据加载到Apache Pig中。

语法

load语句由两部分组成，用“=”运算符分隔。在左侧，需要提到我们想要存储数据的关系的名称；而在右侧，我们需要定义如何存储数据。下面给出了 Load 运算符的语法。

Relation_name = LOAD 'Input file path' USING function as schema;

说明:

relation_name - 我们必须提到要存储数据的关系。
Input file path - 我们必须提到存储文件的HDFS目录。（在MapReduce模式下）
function - 我们必须从Apache Pig提供的一组加载函数中选择一个函数（ BinStorage，JsonLoader，PigStorage，TextLoader ）。
Schema - 我们必须定义数据的模式，可以定义所需的模式如下 -

(column1 : data type, column2 : data type, column3 : data type);

注意:我们加载数据而不指定模式。在这种情况下，列将被寻址为$01，$02，等...（检查）。

例

例如，我们使用 LOAD 命令，在名为学生的模式下在Pig中的 student_data.txt 加载数据。

启动Pig Grunt Shell

首先，打开Linux终端。在MapReduce模式下启动Pig Grunt shell，如下所示。

$ Pig –x mapreduce

它将启动Pig Grunt shell，如下所示。

15/10/01 12:33:37 INFO pig.ExecTypeProvider: Trying ExecType : LOCAL
15/10/01 12:33:37 INFO pig.ExecTypeProvider: Trying ExecType : MAPREDUCE
15/10/01 12:33:37 INFO pig.ExecTypeProvider: Picked MAPREDUCE as the ExecType

2015-10-01 12:33:38,080 [main] INFO  org.apache.pig.Main - Apache Pig version 0.15.0 (r1682971) compiled Jun 01 2015, 11:44:35
2015-10-01 12:33:38,080 [main] INFO  org.apache.pig.Main - Logging error messages to: /home/Hadoop/pig_1443683018078.log
2015-10-01 12:33:38,242 [main] INFO  org.apache.pig.impl.util.Utils - Default bootup file /home/Hadoop/.pigbootup not found
  
2015-10-01 12:33:39,630 [main]
INFO org.apache.pig.backend.hadoop.executionengine.HExecutionEngine - Connecting to hadoop file system at: hdfs://localhost:9000
 
grunt>

执行Load语句

现在，通过在Grunt shell中执行以下Pig Latin语句，将文件 student_data.txt 中的数据加载到Pig中。

grunt> student = LOAD 'hdfs://localhost:9000/pig_data/student_data.txt' 
   USING PigStorage(',')
   as ( id:int, firstname:chararray, lastname:chararray, phone:chararray, 
   city:chararray );

以下是对上述说明的描述。

Relation name

我们已将数据存储在学生（student）模式中。

Input file path

我们从HDFS的/pig_data/目录中的 student_data.txt 文件读取数据。

Storage function

我们使用了 PigStorage() 函数，将数据加载并存储为结构化文本文件。它采用分隔符，使用元组的每个实体作为参数分隔。默认情况下，它以“\t"作为参数。

schema

我们已经使用以下模式存储了数据。

column	id	名字	姓氏	电话号码	城市
datatype	int	char array	char array	char array	char array

注意: Load语句会简单地将数据加载到Pig的指定的关系中。要验证Load语句的执行情况，必须使用Diagnostic运算符，这将在后续的章节中讨论。

Apache Pig 存储数据

在上一章中，我们学习了如何将数据加载到Apache Pig中。你可以使用 store 运算符将加载的数据存储在文件系统中，本章介绍如何使用 Store 运算符在Apache Pig中存储数据。

语法

下面给出了Store语句的语法。

STORE Relation_name INTO ' required_directory_path ' [USING function];

例

假设我们在HDFS中有一个包含以下内容的文件 student_data.txt 。

001,Rajiv,Reddy,9848022337,Hyderabad
002,siddarth,Battacharya,9848022338,Kolkata
003,Rajesh,Khanna,9848022339,Delhi
004,Preethi,Agarwal,9848022330,Pune
005,Trupthi,Mohanthy,9848022336,Bhuwaneshwar
006,Archana,Mishra,9848022335,Chennai.

使用LOAD运算符将它读入关系 student ，如下所示。

grunt> student = LOAD 'hdfs://localhost:9000/pig_data/student_data.txt' 
   USING PigStorage(',')
   as ( id:int, firstname:chararray, lastname:chararray, phone:chararray, 
   city:chararray );

现在，让我们将关系存储在HDFS目录“/pig_Output/"中，如下所示。

grunt> STORE student INTO ' hdfs://localhost:9000/pig_Output/ ' USING PigStorage (',');

输出

执行 store 语句后，将获得以下输出。使用指定的名称创建目录，并将数据存储在其中。

2015-10-05 13:05:05,429 [main] INFO  org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.
MapReduceLau ncher - 100% complete
2015-10-05 13:05:05,429 [main] INFO  org.apache.pig.tools.pigstats.mapreduce.SimplePigStats - 
Script Statistics:
   
HadoopVersion    PigVersion    UserId    StartedAt             FinishedAt             Features 
2.6.0            0.15.0        Hadoop    2015-10-0 13:03:03    2015-10-05 13:05:05    UNKNOWN  
Success!  
Job Stats (time in seconds): 
JobId          Maps    Reduces    MaxMapTime    MinMapTime    AvgMapTime    MedianMapTime    
job_14459_06    1        0           n/a           n/a           n/a           n/a
MaxReduceTime    MinReduceTime    AvgReduceTime    MedianReducetime    Alias    Feature   
     0                 0                0                0             student  MAP_ONLY 
OutPut folder
hdfs://localhost:9000/pig_Output/ 
 
Input(s): Successfully read 0 records from: "hdfs://localhost:9000/pig_data/student_data.txt"  
Output(s): Successfully stored 0 records in: "hdfs://localhost:9000/pig_Output"  
Counters:
Total records written : 0
Total bytes written : 0
Spillable Memory Manager spill count : 0 
Total bags proactively spilled: 0
Total records proactively spilled: 0
  
Job DAG: job_1443519499159_0006
  
2015-10-05 13:06:06,192 [main] INFO  org.apache.pig.backend.hadoop.executionengine
.mapReduceLayer.MapReduceLau ncher - Success!

验证

你可以如下所示验证存储的数据。

步骤1

首先，使用 ls 命令列出名为 pig_output 的目录中的文件，如下所示。

hdfs dfs -ls 'hdfs://localhost:9000/pig_Output/'
Found 2 items
rw-r--r-   1 Hadoop supergroup          0 2015-10-05 13:03 hdfs://localhost:9000/pig_Output/_SUCCESS
rw-r--r-   1 Hadoop supergroup        224 2015-10-05 13:03 hdfs://localhost:9000/pig_Output/part-m-00000

可以观察到在执行 store 语句后创建了两个文件。

步骤2

使用 cat 命令，列出名为 part-m-00000 的文件的内容，如下所示。

$ hdfs dfs -cat 'hdfs://localhost:9000/pig_Output/part-m-00000' 
1,Rajiv,Reddy,9848022337,Hyderabad
2,siddarth,Battacharya,9848022338,Kolkata
3,Rajesh,Khanna,9848022339,Delhi
4,Preethi,Agarwal,9848022330,Pune
5,Trupthi,Mohanthy,9848022336,Bhuwaneshwar
6,Archana,Mishra,9848022335,Chennai