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等工作流调度工具中执行交互式分析。