管理Spark Thrift Server会话

使用Python连接Spark Thrift Server

1. 执行以下命令，安装PyHive和Thrift包。

   pip install pyhive thrift

2. 编写Python脚本，连接Spark Thrift Server。

   以下是一个Python脚本示例，展示如何连接到Hive并显示数据库列表。根据网络环境的不同（外网或内网），您可以选择适合的连接方式。

   使用外网域名连接

   from pyhive import hive
   
   if __name__ == '__main__':
       # 替换<endpoint>、<username>和<token>为您的实际信息。
       cursor = hive.connect('<endpoint>', port=443, scheme='https', username='<username>', password='<token>').cursor()
       cursor.execute('show databases')
       print(cursor.fetchall())
       cursor.close()

   使用内网域名连接

   from pyhive import hive
   
   if __name__ == '__main__':
       # 替换<endpoint>、<username>和<token>为您的实际信息。
       cursor = hive.connect('<endpoint>', port=80, scheme='http', username='<username>', password='<token>').cursor()
       cursor.execute('show databases')
       print(cursor.fetchall())
       cursor.close()

使用Java连接Spark Thrift Server

1. 请在您的pom.xml中引入以下Maven依赖。

   <dependencies>
           <dependency>
               <groupId>org.apache.hadoop</groupId>
               <artifactId>hadoop-common</artifactId>
               <version>3.0.0</version>
           </dependency>
           <dependency>
               <groupId>org.apache.hive</groupId>
               <artifactId>hive-jdbc</artifactId>
               <version>2.1.0</version>
           </dependency>
       </dependencies>
   

   说明

   当前Serverless Spark内置的Hive版本为2.x，因此仅支持hive-jdbc 2.x版本。

2. 编写Java代码，连接Spark Thrift Server。

   以下是一个Sample Java代码，用于连接到Spark Thrift Server，并查询数据库列表。

   使用外网域名连接

   import java.sql.Connection;
   import java.sql.DriverManager;
   import java.sql.ResultSet;
   import java.sql.ResultSetMetaData;
   import org.apache.hive.jdbc.HiveStatement;
   
   public class Main {
       public static void main(String[] args) throws Exception {
           String url = "jdbc:hive2://<endpoint>:443/;transportMode=http;httpPath=cliservice/token/<token>";
           Class.forName("org.apache.hive.jdbc.HiveDriver");
           Connection conn = DriverManager.getConnection(url);
           HiveStatement stmt = (HiveStatement) conn.createStatement();
   
           String sql = "show databases";
           System.out.println("Running " + sql);
           ResultSet res = stmt.executeQuery(sql);
   
           ResultSetMetaData md = res.getMetaData();
           String[] columns = new String[md.getColumnCount()];
           for (int i = 0; i < columns.length; i++) {
               columns[i] = md.getColumnName(i + 1);
           }
           while (res.next()) {
               System.out.print("Row " + res.getRow() + "=[");
               for (int i = 0; i < columns.length; i++) {
                   if (i != 0) {
                       System.out.print(", ");
                   }
                   System.out.print(columns[i] + "='" + res.getObject(i + 1) + "'");
               }
               System.out.println(")]");
           }
           
           conn.close();
       }
   }

   使用内网域名连接

   import java.sql.Connection;
   import java.sql.DriverManager;
   import java.sql.ResultSet;
   import java.sql.ResultSetMetaData;
   import org.apache.hive.jdbc.HiveStatement;
   
   public class Main {
       public static void main(String[] args) throws Exception {
           String url = "jdbc:hive2://<endpoint>:80/;transportMode=http;httpPath=cliservice/token/<token>";
           Class.forName("org.apache.hive.jdbc.HiveDriver");
           Connection conn = DriverManager.getConnection(url);
           HiveStatement stmt = (HiveStatement) conn.createStatement();
   
           String sql = "show databases";
           System.out.println("Running " + sql);
           ResultSet res = stmt.executeQuery(sql);
   
           ResultSetMetaData md = res.getMetaData();
           String[] columns = new String[md.getColumnCount()];
           for (int i = 0; i < columns.length; i++) {
               columns[i] = md.getColumnName(i + 1);
           }
           while (res.next()) {
               System.out.print("Row " + res.getRow() + "=[");
               for (int i = 0; i < columns.length; i++) {
                   if (i != 0) {
                       System.out.print(", ");
                   }
                   System.out.print(columns[i] + "='" + res.getObject(i + 1) + "'");
               }
               System.out.println(")]");
           }
           
           conn.close();
       }
   }

通过Spark Beeline连接Spark Thrift Server

• 如果您使用的是自建集群，需先进入Spark的bin目录，然后使用beeline连接Spark Thrift Server。

  使用外网域名连接

  cd /opt/apps/SPARK3/spark-3.4.2-hadoop3.2-1.0.3/bin/
  
  ./beeline -u "jdbc:hive2://<endpoint>:443/;transportMode=http;httpPath=cliservice/token/<token>"

  使用内网域名连接

  cd /opt/apps/SPARK3/spark-3.4.2-hadoop3.2-1.0.3/bin/
  
  ./beeline -u "jdbc:hive2://<endpoint>:80/;transportMode=http;httpPath=cliservice/token/<token>"

  说明

  代码中的/opt/apps/SPARK3/spark-3.4.2-hadoop3.2-1.0.3是以EMR on ECS 集群的Spark安装路径为例，实际应根据客户端的Spark安装路径进行相应调整。如果您不确定Spark的安装路径，可以通过env | grep SPARK_HOME命令查找。

• 如果您使用的是EMR on ECS 集群，可以直接使用Spark Beeline客户端连接到Spark Thrift Server。

  使用外网域名连接

  spark-beeline -u "jdbc:hive2://<endpoint>:443/;transportMode=http;httpPath=cliservice/token/<token>"

  使用内网域名连接

  spark-beeline -u "jdbc:hive2://<endpoint>:80/;transportMode=http;httpPath=cliservice/token/<token>"

在使用Hive Beeline连接Serverless Spark Thrift Server时，如果出现以下报错，通常是由于Hive Beeline的版本与Spark Thrift Server不兼容导致。因此，建议使用Hive 2.x版本的Beeline。

24/08/22 15:09:11 [main]: ERROR jdbc.HiveConnection: Error opening session
org.apache.thrift.transport.TTransportException: HTTP Response code: 404

配置Apache Superset以连接Spark Thrift Server

Apache Superset是一个现代数据探索和可视化平台，具有丰富的从简单的折线图到高度详细的地理空间图表的图表形态。更多Superset信息，请参见Superset。

1. 安装依赖。

   请确保您已经安装了0.20.0版本的 thrift 包。如未安装，您可以使用以下命令安装。

   pip install thrift==0.20.0

2. 启动Superset，进入Superset界面。

   更多启动操作信息，请参见Superset文档。

3. 在页面右上角单击DATABASE，进入Connect a database页面。

4. 在Connect a database页面，选择Apache Spark SQL。

   

5. 填写连接字符串，然后配置相关数据源参数。

   使用外网域名连接

   hive+https://<username>:<token>@<endpoint>:443/<db_name>

   使用内网域名连接

   hive+http://<username>:<token>@<endpoint>:80/<db_name>

6. 单击FINISH，以确认成功连接和验证。

配置Hue以连接Spark Thrift Server

Hue是一个流行的开源Web界面，可用于与Hadoop生态系统进行交互。关于Hue的更多介绍，请参见Hue官方文档。

1. 安装依赖。

   请确保您已经安装了0.20.0版本的 thrift 包。如未安装，您可以使用以下命令安装。

   pip install thrift==0.20.0

2. 在Hue的配置文件中添加Spark SQL连接串。

   请找到Hue的配置文件（通常位于/etc/hue/hue.conf），并在文件中添加以下内容。

   使用外网域名连接

   [[[sparksql]]]
        name = Spark Sql
        interface=sqlalchemy
        options='{"url": "hive+https://<username>:<token>@<endpoint>:443/"}'

   使用内网域名连接

   [[[sparksql]]]
        name = Spark Sql
        interface=sqlalchemy
        options='{"url": "hive+http://<username>:<token>@<endpoint>:80/"}'

3. 重启Hue。

   修改配置后，您需要执行以下命令重启Hue服务以使更改生效。

   sudo service hue restart

4. 验证连接。

   成功重启后，访问Hue界面，找到Spark SQL选项。如果配置正确，您应能够成功连接到Spark Thrift Server并执行SQL查询。

   

使用DataGrip连接Spark Thrift Server

DataGrip是面向开发人员的数据库管理环境，旨在便捷地进行数据库的查询、创建和管理。数据库可运行于本地、服务器或云端。如需了解更多关于DataGrip的信息，请参见DataGrip。

1. 安装DataGrip，详情请参见Install DataGrip。

   本文示例中的DataGrip版本为2025.1.2。

2. 打开DataGrip客户端，进入DataGrip界面。

3. 创建项目。

   1. 单击，选择New。

      

   2. 在New Project对话框中，输入项目名，例如Spark，单击OK。

4. 单击Database Explorer菜单栏的 图标。选择Data Source > Other > Apache Spark。

   

5. 在Data Sources and Drivers对话框，配置如下参数。

   

   页签	参数	说明
   General	Name	自定义的连接名称。例如，spark_thrift_server。
   	Authentication	选择鉴权方式。本文选择的No auth。 在生产环境中，请选择User & Password，确保只有授权用户能够提交SQL任务，提高系统的安全性。
   	Driver	单击Apache Spark，然后单击Go to Driver ，确认Driver版本为ver. 1.2.2。 说明 由于当前Serverless Spark引擎的版本为3.x，为确保系统的稳定性和功能的兼容性，Driver版本必须选择1.2.2。
   	URL	连接Spark Thrift Server的URL。根据网络环境的不同（外网或内网），您可以选择适合的连接方式。 使用外网域名连接 jdbc:hive2://<endpoint>:443/;transportMode=http;httpPath=cliservice/token/<token> 使用内网域名连接 jdbc:hive2://<endpoint>:80/;transportMode=http;httpPath=cliservice/token/<token>
   Options	Run keep-alive query	该参数为可选配置。勾选该参数可防止超时自动断开。

6. 单击Test Connection，确认数据源配置成功。

   

7. 单击OK，完成配置。

8. 使用DataGrip管理Spark Thrift Server。

   DataGrip成功连接Spark Thrift Server后，您可以进行数据开发，更多信息请参见DataGrip帮助文档。

   例如，您可以在创建的连接下，在目标表上右键，选择New，并在右侧SQL编辑器中编写并运行SQL脚本，即可查看表数据信息。

   

使用Redash连接Spark Thrift Server

Redash是一款开源的BI工具，提供了基于Web的数据库查询和数据可视化功能。如需了解更多关于Redash的信息，请参见Redash官方文档。

1. 安装Redash，详情请参见Redash官方文档。

2. 安装依赖。

   请确保您已经安装了0.20.0版本的thrift包。如未安装，您可以使用以下命令安装。

   pip install thrift==0.20.0

3. 登录Redash。

4. 在左侧导航栏单击Settings，并在Data Sources页签下单击+New Data Source。

5. 在弹出的对话框中配置以下参数，然后单击Create。

   

   参数		说明
   Type Selection		数据源类型。在搜索框中查找并选择Hive（HTTP）。
   Configuration	Name	数据源名称。您可以自定义。
   	Host	Spark Thrift Server的Endpoint地址。 您可以在总览页签获取的Endpoint(外网)或Endpoint(内网)信息。
   	Port	如果使用外网域名访问，端口号为443。 如果使用内网域名访问，端口号为80。
   	HTTP Path	固定填写为/cliservice。
   	Username	用户名。可以任意填写，例如 root。
   	Password	填写您创建的Token信息。
   	HTTP Scheme	如果使用外网域名访问，填写为https。 如果使用内网域名访问，填写为http。

6. 在页面上方选择，您可以在页面的编辑框中编写SQL语句。

   

使用dbt连接Spark Thrift Server

dbt（data build tool）是一个数据转换工具，它允许数据分析师和工程师使用SQL编写数据转换逻辑，并以软件工程的最佳实践（如版本控制、测试、文档）来管理和部署这些转换。如需了解更多关于dbt的信息，请参见dbt官方文档。

1. 安装dbt。

   pip install dbt-spark

   如果您需要使用Hive连接器，也可以安装：

   pip install dbt-spark[HIVE]

2. 创建dbt项目。

   dbt init my_spark_project
   cd my_spark_project

3. 配置dbt profile。

   在~/.dbt/profiles.yml文件中配置Spark Thrift Server连接信息：

   使用外网域名连接

   my_spark_project:
     target: dev
     outputs:
       dev:
         type: spark
         method: thrift
         host: <endpoint>
         port: 443
         user: <username>
         password: <token>
         schema: default
         connect_retries: 5
         connect_timeout: 60
         retry_all: true
         use_ssl: true
         server_side_parameters:
           "hive.exec.dynamic.partition": "true"
           "hive.exec.dynamic.partition.mode": "nonstrict"

   使用内网域名连接

   my_spark_project:
     target: dev
     outputs:
       dev:
         type: spark
         method: thrift
         host: <endpoint>
         port: 80
         user: <username>
         password: <token>
         schema: default
         connect_retries: 5
         connect_timeout: 60
         retry_all: true
         use_ssl: false
         server_side_parameters:
           "hive.exec.dynamic.partition": "true"
           "hive.exec.dynamic.partition.mode": "nonstrict"

4. 测试连接。

   dbt debug

   如果连接成功，您将看到类似以下的输出：

   Connection test: [OK connection ok]

5. 创建dbt模型。

   在models/目录下创建SQL文件，例如models/example_model.sql：

   {{ config(materialized='table') }} select col1, col2, current_timestamp() as created_at from {{ ref('source_table') }} where col1 is not null

6. 运行dbt项目。

   # 运行所有模型
   dbt run
   
   # 运行特定模型
   dbt run --models example_model
   
   # 运行测试
   dbt test
   
   # 生成文档
   dbt docs generate
   dbt docs serve

7. 配置源表和测试。

   在models/schema.yml文件中定义源表和测试：

   version: 2
   
   sources:
     - name: raw_data
       description: "Raw data from source systems"
       tables:
         - name: source_table
           description: "Source table containing raw data"
           columns:
             - name: col1
               description: "Primary identifier"
               tests:
                 - not_null
                 - unique
             - name: col2
               description: "Data column"
   
   models:
     - name: example_model
       description: "Transformed data model"
       columns:
         - name: col1
           description: "Primary identifier"
           tests:
             - not_null
             - unique
         - name: col2
           description: "Processed data column"
         - name: created_at
           description: "Record creation timestamp"

注意事项：

• 确保您的dbt版本与Spark Thrift Server兼容。建议使用dbt-spark 1.3.0或更高版本。

• 在生产环境中，建议将敏感信息（如Token）通过环境变量进行配置，而不是直接写在配置文件中。

• 如果遇到连接超时问题，可以适当调整connect_timeout和connect_retries参数。

• 对于大型数据集，建议使用增量模型以提高性能：

{{ config(
    materialized='incremental',
    unique_key='id',
    incremental_strategy='merge'
) }} select * from source_table {% if is_incremental() %} where updated_at > (select max(updated_at) from {{ this }}) {% endif %}

通过以上配置，您就可以使用dbt连接到Spark Thrift Server，并利用dbt的强大功能进行数据转换和管理。