AnalyticDB：Spark SQL互動式分析

控制台

1. 登入雲原生資料倉儲AnalyticDB MySQL控制台，在左上方選擇叢集所在地區。在左側導覽列，單擊集群清單，然後單擊目的地組群ID。

2. 在左側導覽列單擊作業開發 > SQL開發。

3. 選擇Spark引擎和建立的Spark Interactive型資源群組，執行如下Spark SQL：

   SHOW DATABASES;

SDK調用

通過調用SDK的方法執行Spark SQL時，查詢結果會以檔案的形式寫入指定的OSS中，後續您可以在OSS控制台上查詢資料，或將結果檔案下載到本地查看。下文以Python語言調用SDK為例。

1. 執行以下語句，安裝SDK。

   pip install alibabacloud-adb20211201

2. 依次執行以下語句，安裝環境依賴。

   pip install oss2
   pip install loguru

3. 串連並執行Spark SQL。

   # coding: utf-8
   import csv
   import json
   import time
   from io import StringIO
   
   import oss2
   from alibabacloud_adb20211201.client import Client
   from alibabacloud_adb20211201.models import ExecuteSparkWarehouseBatchSQLRequest, ExecuteSparkWarehouseBatchSQLResponse, \
       GetSparkWarehouseBatchSQLRequest, GetSparkWarehouseBatchSQLResponse, \
       ListSparkWarehouseBatchSQLRequest, CancelSparkWarehouseBatchSQLRequest, ListSparkWarehouseBatchSQLResponse
   from alibabacloud_tea_openapi.models import Config
   from loguru import logger
   
   def build_sql_config(oss_location,
                        spark_sql_runtime_config: dict = None,
                        file_format = "CSV",
                        output_partitions = 1,
                        sep = "|"):
       """
       構建ADB SQL執行的配置
       :param oss_location: OSS存放SQL執行結果的路徑
       :param spark_sql_runtime_config: Spark SQL社區的原生配置
       :param file_format: SQL執行結果的檔案格式, 預設為CSV
       :param output_partitions: SQL執行結果的分區數, 如果需要輸出大量結果, 必須增加輸出時的分區數避免單個檔案過大
       :param sep: CSV檔案的分隔字元, 非CSV檔案忽略
       :return: SQL執行的配置
       """
   
       if oss_location is None:
           raise ValueError("oss_location is required")
   
       if not oss_location.startswith("oss://"):
           raise ValueError("oss_location must start with oss://")
   
       if file_format != "CSV" and file_format != "PARQUET" and file_format != "ORC" and file_format != "JSON":
           raise ValueError("file_format must be CSV, PARQUET, ORC or JSON")
   
       runtime_config =  {
           # sql output config
           "spark.adb.sqlOutputFormat": file_format,
           "spark.adb.sqlOutputPartitions": output_partitions,
           "spark.adb.sqlOutputLocation": oss_location,
           # csv config
           "sep": sep
       }
   
       if spark_sql_runtime_config:
           runtime_config.update(spark_sql_runtime_config)
       return runtime_config
   
   
   def execute_sql(client: Client, dbcluster_id: str,
                   resource_group_name: str,
                   query: str,
                   limit = 10000,
                   runtime_config: dict = None,
                   schema="default"
                   ):
       """
       在Spark Interactive型資源群組中執行SQL
       :param client: 阿里雲用戶端
       :param dbcluster_id: 叢集的ID
       :param resource_group_name: 叢集的資源群組,要求必須是Spark Interactive資源群組
       :param schema: SQL執行的預設的資料庫名稱, 不填寫為default
       :param limit: SQL執行的結果的行數限制
       :param query: 執行的SQL語句, 使用分號分隔多個SQL語句
       :return:
       """
       # 組裝請求體
       req = ExecuteSparkWarehouseBatchSQLRequest()
       # 叢集ID
       req.dbcluster_id = dbcluster_id
       # 資源群組名稱
       req.resource_group_name = resource_group_name
       # SQL執行逾時時間
       req.execute_time_limit_in_seconds = 3600
       # 執行SQL的資料庫名稱
       req.schema = schema
       # SQL業務代碼
       req.query = query
       # 返回結果行數
       req.execute_result_limit = limit
   
       if runtime_config:
       # SQL執行的配置
           req.runtime_config = json.dumps(runtime_config) 
       # 執行SQL並擷取query_id
       resp: ExecuteSparkWarehouseBatchSQLResponse = client.execute_spark_warehouse_batch_sql(req)
       logger.info("Query execute submitted: {}", resp.body.data.query_id)
       return resp.body.data.query_id
   
   
   def get_query_state(client, query_id):
       """
       查詢SQL執行的狀態
       :param client: 阿里雲用戶端
       :param query_id: SQL執行的ID
       :return: SQL執行的狀態和結果
       """
       req = GetSparkWarehouseBatchSQLRequest(query_id=query_id)
       resp: GetSparkWarehouseBatchSQLResponse = client.get_spark_warehouse_batch_sql(req)
       logger.info("Query state: {}", resp.body.data.query_state)
       return resp.body.data.query_state, resp
   
   def list_history_query(client, db_cluster, resource_group_name, page_num):
       """
       查詢Spark Interactive資源群組中執行的SQL的歷史
       :param client: 阿里雲用戶端
       :param db_cluster: 叢集的ID
       :param resource_group_name: 資源群組名稱
       :param page_num: 分頁查詢的頁數
       :return: 是否有SQL, 有的話代表可以進入下一頁查詢
       """
       req = ListSparkWarehouseBatchSQLRequest(dbcluster_id=db_cluster,
                                               resource_group_name=resource_group_name,
                                               page_number = page_num)
       resp: ListSparkWarehouseBatchSQLResponse = client.list_spark_warehouse_batch_sql(req)
   
       # 如果沒有查詢到SQL, 返回false. 否則返回true. 預設每頁10條
       if resp.body.data.queries is None:
           return True
   
       # 列印查詢到的SQL
       for query in resp.body.data.queries:
           logger.info("Query ID: {}, State: {}", query.query_id, query.query_state)
       logger.info("Total queries: {}", len(resp.body.data.queries))
       return len(resp.body.data.queries) < 10
   
   
   def list_csv_files(oss_client, dir):
       for obj in oss_client.list_objects_v2(dir).object_list:
           if obj.key.endswith(".csv"):
               logger.info(f"reading {obj.key}")
               # read oss file content
               csv_content = oss_client.get_object(obj.key).read().decode('utf-8')
               csv_reader = csv.DictReader(StringIO(csv_content))
               # Print the CSV content
               for row in csv_reader:
                   print(row)
   
   
   if __name__ == '__main__':
       logger.info("ADB Spark Batch SQL Demo")
   
       # AccessKey ID，請填寫實際值
       _ak = "LTAI****************"
       # AccessKey Serect，請填寫實際值
       _sk = "yourAccessKeySecret"
       # 地區ID，請填寫實際值
       _region= "cn-shanghai" 
   
   
       # 叢集ID，請填寫實際值
       _db = "amv-uf6485635f****"
       # 資源群組名稱，請填寫實際值
       _rg_name = "testjob" 
   
       # client config
       client_config = Config(
           # Alibaba Cloud AccessKey ID
           access_key_id=_ak,
           # Alibaba Cloud AccessKey Secret
           access_key_secret=_sk,
           # The endpoint of the ADB service
           # adb.ap-southeast-1.aliyuncs.com is the endpoint of the ADB service in the Singapore region
           # adb-vpc.ap-southeast-1.aliyuncs.com used in the VPC scenario
           endpoint=f"adb.{_region}.aliyuncs.com"
       )
   
       # 建立阿里雲用戶端
       _client = Client(client_config)
   
       # SQL執行的配置
       _spark_sql_runtime_config = {
           "spark.sql.shuffle.partitions": 1000,
           "spark.sql.autoBroadcastJoinThreshold": 104857600,
           "spark.sql.sources.partitionOverwriteMode": "dynamic",
           "spark.sql.sources.partitionOverwriteMode.dynamic": "dynamic"
       }
       _config = build_sql_config(oss_location="oss://testBucketName/sql_result",
                                  spark_sql_runtime_config = _spark_sql_runtime_config)
       # 需要執行的SQL
       _query = """
       SHOW DATABASES;
       SELECT 100;
       """
   
       _query_id = execute_sql(client = _client,
                               dbcluster_id=_db,
                               resource_group_name=_rg_name,
                               query=_query, runtime_config=_config)
       logger.info(f"Run query_id: {_query_id} for SQL {_query}.\n Waiting for result...")
   
       # 等待SQL執行完成
       current_ts = time.time()
       while True:
           query_state, resp = get_query_state(_client, _query_id)
           """
           query_state 有如下幾種狀態:
           - PENDING: 在服務隊列排隊中, 此時Spark Interactive型資源群組正在啟動
           - SUBMITTED: 提交到了Spark Interactive型資源群組
           - RUNNING: SQL正在執行
           - FINISHED: SQL執行完成, 沒有報錯
           - FAILED: SQL執行失敗
           - CANCELED: SQL執行被取消
           """
   
           if query_state == "FINISHED":
               logger.info("query finished success")
               break
           elif query_state == "FAILED":
               # 列印失敗資訊
               logger.error("Error Info: {}", resp.body.data)
               exit(1)
           elif query_state == "CANCELED":
               # 列印取消資訊
               logger.error("query canceled")
               exit(1)
           else:
               time.sleep(2)
               if time.time() - current_ts > 600:
                   logger.error("query timeout")
                   # 執行時間超過10分鐘, 取消SQL執行
                   _client.cancel_spark_warehouse_batch_sql(CancelSparkWarehouseBatchSQLRequest(query_id=_query_id))
                   exit(1)
   
       # 一個Query可以包含多個語句, 列舉所有的語句
       for stmt in resp.body.data.statements:
           logger.info(
               f"statement_id: {stmt.statement_id}, result location: {stmt.result_uri}")
           # 查看結果的範例程式碼
           _bucket = stmt.result_uri.split("oss://")[1].split("/")[0]
           _dir = stmt.result_uri.replace(f"oss://{_bucket}/", "").replace("//", "/")
           oss_client = oss2.Bucket(oss2.Auth(client_config.access_key_id, client_config.access_key_secret),
                                    f"oss-{_region}.aliyuncs.com", _bucket)
           list_csv_files(oss_client, _dir)
   
       # 查詢在Spark Interactive資源群組中執行的所有的SQL,可以分頁查詢
       logger.info("List all history query")
       page_num = 1
       no_more_page = list_history_query(_client, _db, _rg_name, page_num)
       while no_more_page:
           logger.info(f"List page {page_num}")
           page_num += 1
           no_more_page = list_history_query(_client, _db, _rg_name, page_num)
   

   參數說明：

   • _ak：阿里雲帳號或具備AnalyticDB for MySQL存取權限的RAM使用者的AccessKey ID。如何擷取AccessKey ID和AccessKey Secret，請參見帳號與許可權。

   • _sk：阿里雲帳號或具備AnalyticDB for MySQL存取權限的RAM使用者的AccessKey Secret。如何擷取AccessKey ID和AccessKey Secret，請參見帳號與許可權。

   • region：AnalyticDB for MySQL叢集所屬地區ID。

   • _db：AnalyticDB for MySQL叢集ID。

   • _rg_name：Spark Interactive型資源群組的名稱。

   • oss_location（可選）：查詢結果檔案儲存的OSS路徑。

     若不填寫該參數，您僅可以在作業開發 > Spark Jar開發頁面，對應SQL查詢語句的日誌中，查看到5行資料。

應用程式

用戶端

除了在本文中詳細介紹的Beeline、DBeaver、DBVisualizer、Datagrip用戶端外，您還可以在Airflow、Azkaban、DolphinScheduler等工作流程調度工具中執行互動式分析。