通过Spark SQL Engine开发Spark SQL作业 - 云原生数据仓库AnalyticDB

当您需要实时分析数据或通过JDBC协议开发Spark SQL作业时，可以通过云原生数据仓库 AnalyticDB MySQL 版的Spark Distribution SQL Engine开发Spark SQL作业。通过Spark Distribution SQL Engine可以更方便地利用SQL分析、处理和查询数据，提升SQL执行效率。本文主要介绍如何启动、连接Spark Distribution SQL Engine。

功能介绍

AnalyticDB for MySQL Spark提供Spark开源原生能力Distribution SQL Engine的托管服务。Spark Distribution SQL Engine是Spark内置的分布式SQL执行引擎，支持JDBC协议。任何支持标准Hive ThriftServer2的客户端均可以通过JDBC连接Spark Distribution SQL Engine。

您可以开启多个Spark Distribution SQL Engine，多个Spark Distribution SQL Engine之间执行环境相互隔离，且都具有独立的连接地址，但是Spark Distribution SQL Engine之间会共享带宽，因此不适用于大量数据的导入。更多关于Spark Distribution SQL Engine的信息，请参见Distributed SQL Engine。

前提条件

集群的产品系列为湖仓版。
集群与OSS存储空间位于相同地域。
已创建Job型资源组。具体操作，请参见新建资源组。
已创建数据库账号。
- 如果是通过阿里云账号访问，只需创建高权限账号。具体操作，请参见创建高权限账号。
- 如果是通过RAM用户访问，需要创建高权限账号和普通账号并且将RAM用户绑定到普通账号上。具体操作，请参见创建数据库账号和绑定或解绑RAM用户与数据库账号。
已安装Java 8开发环境，以运行Beeline客户端和Java应用。
已将客户端IP地址添加至AnalyticDB for MySQL集群白名单中。具体操作，请参见白名单。

启动Spark Distribution SQL Engine

登录云原生数据仓库AnalyticDB MySQL控制台，在左上角选择集群所在地域。在左侧导航栏，单击集群列表，在湖仓版页签下，单击目标集群ID。
在左侧导航栏，单击作业开发 > Spark Jar开发。
在编辑器窗口上方，选择Job型资源组，作业类型选择SQLEngine。

在编辑器中输入以下作业内容。

-- Here is just an example of start Spark SQLEngine. Modify the content and run your spark program.
CONF spark.adb.eni.enabled=true;
CONF spark.adb.eni.vswitchId=vsw-bp1wh6lp1abs3fi0****;
CONF spark.adb.eni.securityGroupId=sg-bp15ncw8v0624xe****;
CONF spark.driver.resourceSpec=medium;
CONF spark.executor.instances=1;
CONF spark.executor.resourceSpec=small;
CONF spark.app.name=Spark SQLEngine;
CONF spark.sql.hive.metastore.version=adb;
CONF spark.kubernetes.driverEnv.HIVE_SERVER2_USER=AdbSpark14****;
CONF spark.kubernetes.driverEnv.HIVE_SERVER2_PASSWORD=Spark23****;
CONF spark.adb.sessionTTLSeconds=1200;

参数说明：

参数	是否必填	说明
spark.adb.eni.enabled	条件必填	是否开启弹性网卡（ENI）。通过IP地址连接Spark Distribution SQL Engine时，需配置为true。通过域名连接Spark Distribution SQL Engine时，无需配置。
spark.adb.eni.vswitchId	条件必填	弹性网卡的交换机ID。通过IP地址连接Spark Distribution SQL Engine时，则需要配置为客户端所属VPC网络的交换机ID。通过域名连接Spark Distribution SQL Engine时，无需配置。
spark.adb.eni.securityGroupId	条件必填	弹性网卡的安全组ID。通过IP地址连接Spark Distribution SQL Engine时，则需要配置为客户端所属VPC网络的安全组ID。通过域名连接Spark Distribution SQL Engine时，无需配置。
spark.app.name	是	Spark应用名称。
spark.sql.hive.metastore.version	是	指定采用元数据服务版本，支持如下配置： adb：连接AnalyticDB for MySQL中的元数据信息。 <hive_version>：指定Hive MetaStore的版本。说明 Spark支持的Hive版本，请参见Spark Configuration。如需要连接自建的Hive Metastore，可以采用社区Spark的标准配置方式。请参见Spark Configuration。
spark.kubernetes.driverEnv.HIVE_SERVER2_USER	否	Spark Distribution SQL Engine服务的用户名。用户名需包含大写字母、小写字母和数字，长度无限制。
spark.kubernetes.driverEnv.HIVE_SERVER2_PASSWORD	否	Spark Distribution SQL Engine服务的密码。密码需包含大写字母、小写字母和数字，长度无限制。
spark.adb.sessionTTLSeconds	否	Spark Distribution SQL Engine销毁时间。单位为秒（s），默认值为1200秒。即最后一个SQL代码块执行完毕，1200秒后Spark Distribution SQL Engine服务会自动销毁。说明 Spark Distribution SQL Engine销毁后，Spark Jar作业的状态为已完成。

其他参数，请参见Spark应用配置参数说明。

单击立即执行。
重要
您可在应用列表区域查看任务状态，仅当任务状态为运行中时，可以连接Spark Distribution SQL Engine。

连接Spark Distribution SQL Engine

支持标准Hive ThriftServer的客户端均可以通过IP地址或域名连接Spark Distribution SQL Engine。客户端列表，请参见Hive Client官方文档。

通过IP地址和通过域名连接Spark Distribution SQL Engine的区别如下：

通过IP地址连接Spark Distribution SQL Engine（推荐）：客户端与ENI所属同一VPC，且已开启ENI（spark.adb.eni.vswitchId为true）时，可以通过IP地址连接，此时您需要获取Spark Distribution SQL Engine的IP地址。若您的客户端所在VPC发生改变，您需修改ENI所在的交换机（即spark.adb.eni.vswitchId参数）并重新启动Spark Distribution SQL Engine。
通过域名连接Spark Distribution SQL Engine：客户端与AnalyticDB for MySQL集群所属同一VPC，或需要通过公网连接时，可以通过域名连接，此时您需获取Spark Distribution SQL Engine的域名地址。

步骤一：获取Spark Distribution SQL Engine的连接地址

获取Spark Distribution SQL Engine的IP地址

登录云原生数据仓库AnalyticDB MySQL控制台，在左上角选择集群所在地域。在左侧导航栏，单击集群列表，在湖仓版页签下，单击目标集群ID。
在左侧导航栏，单击作业开发 > Spark Jar开发。
在应用列表页签中，单击目标Spark应用操作列的日志，获取Spark Distribution SQL Engine的IP地址。
```
jdbc:hive2://172.XX.XX.XX:10001/<db>;transportMode=http;httpPath=s202401301607hz**********.cliservice
```

获取Spark Distribution SQL Engine的域名地址

登录云原生数据仓库AnalyticDB MySQL控制台，在左上角选择集群所在地域。在左侧导航栏，单击集群列表，在湖仓版页签下，单击目标集群ID。
在左侧导航栏，单击作业开发 > Spark Jar开发。
在应用列表页签中，单击目标Spark应用操作列的详情，获取Spark Distribution SQL Engine的域名地址。
- 通过公网连接Spark Distribution SQL Engine时，需使用Spark JDBC Public Connect URL对应的地址。
- 通过VPC连接Spark Distribution SQL Engine时，需使用Spark JDBC VPC Connect URL对应的地址。
重要
- 首次启动Spark Distribution SQL Engine后，会进行网络初始化，初始化过程持续时间大约为1分钟，初始化完成后才能获取域名地址。
- 创建Spark域名与申请、释放公网地址的操作不能同时进行，因此不能确保每次都成功创建域名。若您多次启动Spark Distribution SQL Engine均无法创建域名时，请提交工单联系技术支持解决。

步骤二：连接Spark Distribution SQL Engine

本文示例以Beeline客户端和Java应用为例，介绍连接Spark Distribution SQL Engine的具体方法。

通过Beeline客户端连接Spark Distribution SQL Engine

连接Spark Distribution SQL Engine。
命令格式如下：
```
!connect <连接地址> <用户名> <密码>
```
- 连接地址：步骤一中获取的Spark Distribution SQL Engine连接地址。
- 用户名：Spark Distribution SQL Engine的用户名。
- 密码：Spark Distribution SQL Engine的密码。
示例：
```
!connect jdbc:hive2://jdbc:hive2://amv-bp1c3em7b2e****-spark.ads.aliyuncs.com:10001/adb_test;transportMode=http;httpPath=s202401301607hz**********.cliservice AdbSpark14**** Spark23****
```
重要
- 启动Spark Distribution SQL Engine时，若设置了spark.sql.hive.metastore.version=adb，则连接地址中的数据库必须为AnalyticDB for MySQL中已存在的数据库，否则会默认使用default 数据库，造成连接失败。
- ThriftServer HTTP Protocol会在网络波动时，出现509拒绝访问报错，且不会自动重试，您需要手动重新执行。
- 自动生成的连接配置参数transportMode=http;httpPath=s202401301607hzdf852**********.cliservice不能修改和删除，若您需要新增其它配置参数，在连接地址末尾添加即可。
返回结果：
```
Connected to: Spark SQL (version 3.2.0)
Driver: Hive JDBC (version 2.3.9)
Transaction isolation: TRANSACTION_REPEATABLE_READ
```
使用Spark Distribution SQL Engine执行Spark SQL。
```
SHOW TABLES;
```

通过Java应用连接Spark Distribution SQL Engine

设置环境变量。

export HIVEURL="jdbc:hive2://jdbc:hive2://amv-bp1c3em7b2e****-spark.ads.aliyuncs.com:10001/adb_test;transportMode=http;httpPath=s202401301607hz**********.cliservice"
export HIVEUSER="AdbSpark14****"
export HIVEPASSWORD="Spark23****"

HIVEURL：步骤一中获取的Spark Distribution SQL Engine连接地址。
HIVEUSER：Spark Distribution SQL Engine的用户名。
HIVEPASSWORD：Spark Distribution SQL Engine的密码。

在pom.xml中配置Maven依赖。示例代码如下：

    <dependency>
      <groupId>org.apache.hive</groupId>
      <artifactId>hive-jdbc</artifactId>
      <version>2.3.9</version>
    </dependency>

建立连接并执行Spark SQL。

package com.aliyun;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;
public class TestHive {
    public static void main(String[] args) throws Exception {
        String hiveUrl = System.getenv("HIVEURL");;
        String username = System.getenv("HIVEUSER");
        String password = System.getenv("HIVEPASSWORD");
        // 已安装驱动
        Class.forName("org.apache.hive.jdbc.HiveDriver");
        // 通过步骤一获取的地址连接Spark Distribution SQL Engine
        String url = String.format(hiveUrl);
        Connection con = DriverManager.getConnection(url, username, password);
        Statement stmt = con.createStatement();
        // 执行show tables命令
        ResultSet tables = stmt.executeQuery("show tables");
        while (tables.next()) {
            System.out.println(tables.getString("tableName"));
        }
        // 查询表数据
        ResultSet rows = stmt.executeQuery("select * from test");
        while (rows.next()) {
            System.out.println(rows.getString(0));
        }
    }
}

重要

启动Spark Distribution SQL Engine时，若设置了spark.sql.hive.metastore.version=adb，则连接地址中的数据库必须为AnalyticDB for MySQL中已存在的数据库，否则会默认使用default 数据库，造成连接失败。
ThriftServer HTTP Protocol会在网络波动时，出现509拒绝访问报错，且不会自动重试，您需要手动重新执行。
自动生成的连接配置参数transportMode=http;httpPath=s202401301607hzdf852**********.cliservice不能修改和删除，若您需要新增其它配置参数，在连接地址末尾添加即可。