E-MapReduce：管理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的強大功能進行資料轉換和管理。