Spark对接Phoenix4.x快速入门-阿里云帮助中心

Phoenix擅长在线简单查询，复杂分析场景不适用。通过分析集群可以加强对Phoenix中数据的分析。当前云HBase支持Phoenix4.x（对应HBase1.x版本）和Phoenix5.x（对应HBase2.x版本）两个版本，分别采用重客户端和轻客户端模式，在分析时使用不同的方式读写Phoenix数据。本文主要介绍通过“数据工作台”使用Spark对接Phoenix4.x版本数据的方法。

前置条件

Spark集群和HBase集群在同一个VPC下。进入Spark分析集群页面，选择“数据库连接”>“连接信息”，查看Spark集群的VPC ID信息。如下图：
Spark分析集群关联HBase集群。进入Spark分析集群页面，选择“关联数据库”>“关联HBase”，支持关联以及取消关联。如下图：

Phoenix已创建表，本实例使用的Phoenix表：us_population,建表语句和内容如下：

#建表语句：
CREATE TABLE IF NOT EXISTS us_population (
state CHAR(2) NOT NULL,
city VARCHAR NOT NULL,
population BIGINT
CONSTRAINT my_pk PRIMARY KEY (state, city));
#插入数据语句：
UPSERT INTO us_population VALUES('NY','New York',8143197);
UPSERT INTO us_population VALUES('CA','Los Angeles',3844829);
UPSERT INTO us_population VALUES('IL','Chicago',2842518);
UPSERT INTO us_population VALUES('TX','Houston',2016582);
UPSERT INTO us_population VALUES('PA','Philadelphia',1463281);
UPSERT INTO us_population VALUES('AZ','Phoenix',1461575);
UPSERT INTO us_population VALUES('TX','San Antonio',1256509);
UPSERT INTO us_population VALUES('CA','San Diego',1255540);
UPSERT INTO us_population VALUES('TX','Dallas',1213825);
UPSERT INTO us_population VALUES('CA','San Jose',912332);

使用“数据工作台”>“作业管理”运行样例

步骤 1：通过“资源管理”上传样例代码Jar包

下载样例代码jar包“spark-examples-0.0.1-SNAPSHOT.jar”到本地目录。

wget https://spark-home.oss-cn-shanghai.aliyuncs.com/spark_example/spark-examples-0.0.1-SNAPSHOT.jar

在“数据工作台”>“资源管理”中添加文件夹“spark_on_phoenix”。上传jar包“spark-examples-0.0.1-SNAPSHOT.jar”到此文件夹。如下图：

步骤 2：通过“作业管理”创建并编辑作业内容

在“数据工作台”>“作业管理”中创建Spark作业，作业内容如下：

--class com.aliyun.spark.phoenix.SparkOnPhoenix4xSparkSession
--driver-memory 1G 
--driver-cores 1
--executor-cores 1
--executor-memory 2G
--num-executors 1
--name spark_on_phoenix
 /spark_on_phoenix/spark-examples-0.0.1-SNAPSHOT.jar
hb-xxx:2181,hb-xxx:2181,hb-xxx:2181 us_population spark_on_phoenix

作业内容参数说明：


参数	说明
hb-xxx:2181,hb-xxx:2181,hb-xxx:2181	HBase集群中的“ZK链接地址（专有网络）”。
us_population	Phoenix的表名。
spark_on_phoenix	Spark中创建映射Phoenix表的表名。

如下图：

步骤 3：通过“作业管理”运行作业并查看结果

作业编辑完成后点击“运行”，选择Spark集群。运行状态会在下侧显示，如图：

运行成功后点击“YarnUI”，翻到最后看到如下结果表明Spark读取Phoenix成功。如下：

+-----+-------+----------+
|STATE|   CITY|POPULATION|
+-----+-------+----------+
|   AZ|Phoenix|   1461575|
+-----+-------+----------+

使用“交互式查询”运行样例

步骤 1：通过“会话管理”创建会话

在“数据工作台”>“会话管理”中点击“创建会话”。填写“会话名称”：spark_on_phoenix, 选择需要执行的Spark集群。点击“确认”后，编辑“会话内容”，内容如下：

--driver-memory 1G 
--driver-cores 1
--executor-cores 1
--executor-memory 2G
--num-executors 1
--name spark_on_phoenix

内容编辑完成后，点击“运行”（此处点击运行是为了下一步骤中可以选择会话“spark_on_phoenix”）。如下图：

步骤 2：通过“交互式查询”创建查询

进入“数据工作台”>“交互式查询”，在“会话列表”下拉框中选择“spark_on_phoenix”，然后点击“新建查询”。填写“查询名称”：spark_on_phoenix，“查询类型”选择“sql”。如下图：

步骤 3：通过“交互式查询”编辑查询

打开上步骤创建的“spark_on_phoenix”，输入如下内容：

CREATE TABLE spark_on_phoenix01 USING org.apache.phoenix.spark
OPTIONS (
'zkUrl' 'hb-xxx:2181,hb-xxx:2181,hb-xxx:2181',
'table' 'us_population'
)

如下图：

建表语句关键字说明：


关键字	说明
spark_on_phoenix01	Spark SQL中创建的表名。
hb-xxx:2181,hb-xxx:2181,hb-xxx:2181	HBase集群中的“ZK链接地址（专有网络）”。
us_population	Phoenix的表名。

步骤 4：通过“交互式查询”运行查询

编辑查询“spark_on_phoenix”完成后，点击“运行”；运行成功后，在“查询内容”中输入“select * from spark_on_phoenix01”，然后点击“运行”验证结果，出现如下图结果表示运行成功。如下图：

小结

本例代码可参考：SparkOnPhoenix4xSparkSession.scala
更多样例代码可参考：Spark样例代码
分析Phoenix的建表DDL语法和Phoenix社区的Spark Plugin一致；查询语法和Spark SQL一致，支持ANSI SQL 2003语法。

注意事项

社区版Phoenix对于时区的处理逻辑有缺陷，阿里云Phoenix修复了该缺陷参考文章。对于老版本phoenix用户(4.x版本)使用Phoenix Spark connector需做如下步骤：

HBase集群小版本升级到最新版本
提交spark作业的时候加如下两个配置项：
```
spark.hadoop.phoenix.timezone.conversion=true
spark.hadoop.phoenix.local.timezone=GMT+8 （运行phoenix重客户端所在的时区）
```
具体配置可以使用spark查询出时区数据校验，对于使用阿里云修复时区之后的phoenix版本不加前述配置，直接使用即可。