Spark对接分析型数据库PostgreSQL版快速入门_Spark ETL&Streaming数据源连接器_X-Pack Spark计算服务

简介

AnalyticDB for PostgreSQL（原HybridDB for PostgreSQL）为您提供简单、快速、经济高效的 PB 级云端数据仓库解决方案。
本文主要介绍通过“数据工作台”使用Spark对接分析型数据库PostgreSQL版的方法。

前置条件

Spark集群和分析型数据库PostgreSQL版在同一个VPC下。
进入Spark分析集群页面，选择“数据库连接”>“连接信息”，查看Spark集群的VPC ID信息。如下图：
然后在阿里云控制台进入“专有网络VPC”，找到Spark集群的VPC ID对应的IPv4网段。如下图：
最后进入PostgreSQL配置白名单，添加上述IPv4网段。如下图：

PostgreSQL数据库已创建表，本实例使用的表名为：test_table，通过如下方法在数据库中创建表：
下载样例代码“common-test-0.0.1-SNAPSHOT-shaded.jar”到ECS集群上的位置为：/opt/jars/test/common-test-0.0.1-SNAPSHOT-shaded.jar
注意：ECS集群需要和数据在同一个VPC下，且数据库配置了ECS的白名单。

wget https://spark-home.oss-cn-shanghai.aliyuncs.com/common_test/common-test-0.0.1-SNAPSHOT-shaded.jar

然后运行：

java -cp /opt/jars/test/common-test-0.0.1-SNAPSHOT-shaded.jar com.aliyun.adb.CreateSampleTable 'jdbc:postgresql://xxx1:3432/postgres' xxx2 xxx3 test_table

参数说明：

参数	说明
xxx1:3432	分析型数据库PostgreSQL版的数据库连接的“内网地址”和“内网端口”。
postgres test_table	分析型数据库PostgreSQL版数据库名和表名。
xxx2 xxx3	分别为分析型数据库PostgreSQL版登陆数据库的用户名和密码。

显示如下结果说明测试表创建成功：

aliyun02   102   10.0
aliyun04   104   10.0
aliyun01   101   10.0
aliyun03   103   10.0
aliyun05   105   10.0

本实例中使用的表结构为：

 "name" varchar(32) NULL,
 "age" int  NULL,
 "score" double precision  NULL

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

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

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

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

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

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

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

--class com.aliyun.spark.adb.SparkOnADBPostgreSQLSparkSession
--jars /spark_on_adb/postgresql-42.2.5.jar
--driver-memory 1G 
--driver-cores 1
--executor-cores 1
--executor-memory 2G
--num-executors 1
--name spark_on_adb
/spark_on_adb/spark-examples-0.0.1-SNAPSHOT.jar
xxx1:3432 postgres public test_table xxx2 xxx3 spark_on_adb_pg01

作业内容参数说明：

参数	说明
xxx1:3432	分析型数据库PostgreSQL版的数据库连接的“内网地址”和“内网端口”。
postgres public test_table	分别为分析型数据库 PostgreSQL版的数据库名、schema（模式）名和表名。
xxx2 xxx3	分别为分析型数据库PostgreSQL版的数据库登陆的用户名和密码。
spark_on_adb_pg01	Spark中创建映射分析型数据库PostgreSQL版数据库表的表名。

如下图：

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

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

运行成功后点击“YarnUI”，翻到最后看到如下结果表明Spark读取分析型数据库PostgreSQL成功。如下：

+--------+----+-----+
|    name| age|score|
+--------+----+-----+
|aliyun01|1001| 10.1|
+--------+----+-----+

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

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

在“数据工作台”>“会话管理”中点击“创建会话”。填写“会话名称”：spark_on_adb, 选择需要执行的Spark集群。

点击“确认”后，编辑“会话内容”，内容如下：

--name spark_on_adb
--driver-memory 1G 
--driver-cores 1
--executor-cores 1
--executor-memory 2G
--num-executors 1
--jars /spark_on_adb/postgresql-42.2.5.jar

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

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

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

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

查询输入如下内容：

CREATE TABLE spark_on_adb_pg02 USING org.apache.spark.sql.jdbc
    options (
    driver 'org.postgresql.Driver',
    url 'jdbc:postgresql://xxx1:3432/postgres',
    dbtable 'public.test_table',
    user 'xxx2',
    password 'xxx3'
)

建表语句关键字说明：

参数	说明
xxx1:3432	分析型数据库PostgreSQL版的数据库连接的“内网地址”和“内网端口”。
postgres public test_table	分别为分析型数据库 PostgreSQL版的数据库名、schema（模式）名和表名。
xxx2 xxx3	分别为分析型数据库PostgreSQL版的数据库登陆的用户名和密码。
spark_on_adb_pg02	Spark中创建映射分析型数据库PostgreSQL版数据库表的表名。

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

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

小结

本例代码可参考：SparkOnADB
更多样例代码可参考：Spark样例代码