AnalyticDB:Interactive analysis with Spark SQL

Console

1. Log on to the AnalyticDB for MySQL console. In the upper-left corner, select a region. In the left navigation pane, click Clusters, then click the cluster ID.

2. In the left navigation pane, choose Job Development > SQL Development.

3. Select the Spark engine and your Spark Interactive resource group, then run:

   SHOW DATABASES;

Python SDK

The SDK submits queries asynchronously. Results are written to an OSS path you specify, and you can download the result file to view it locally.

All examples use ExecuteSparkWarehouseBatchSQLRequest to submit a query, poll for completion via GetSparkWarehouseBatchSQLRequest, then read each statement's result from OSS.

1. Install the SDK and dependencies.

   pip install alibabacloud-adb20211201 oss2 loguru

2. Submit and retrieve Spark SQL results.

   Placeholder	Description
   <your-access-key-id>	AccessKey ID of an Alibaba Cloud account or RAM user with AnalyticDB for MySQL access permissions. See Accounts and permissions.
   <your-access-key-secret>	AccessKey secret for the same account or RAM user.
   <region-id>	Region ID of the cluster, for example, cn-shanghai.
   <cluster-id>	AnalyticDB for MySQL cluster ID, for example, amv-uf6485635f****.
   <resource-group-name>	Name of the Spark Interactive resource group.
   oss://<your-bucket>/sql_result	(Optional) OSS path for result files. If omitted, only the first five rows appear in the query logs on the Job Development > Spark JAR Development page.

   # 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="|",
   ):
       """
       Build the output configuration for a Spark SQL execution.
   
       :param oss_location: OSS path where results are stored. Must start with oss://.
       :param spark_sql_runtime_config: Additional Spark SQL runtime properties.
       :param file_format: Output format — CSV, PARQUET, ORC, or JSON. Default: CSV.
       :param output_partitions: Number of output partitions. Increase this for large result sets
                                 to avoid generating a single oversized file.
       :param sep: Column separator for CSV output. Ignored for non-CSV formats.
       """
       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 not in ("CSV", "PARQUET", "ORC", "JSON"):
           raise ValueError("file_format must be CSV, PARQUET, ORC, or JSON")
   
       runtime_config = {
           "spark.adb.sqlOutputFormat": file_format,
           "spark.adb.sqlOutputPartitions": output_partitions,
           "spark.adb.sqlOutputLocation": oss_location,
           "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",
   ):
       """
       Submit a Spark SQL query to a Spark Interactive resource group.
   
       :param client: Alibaba Cloud SDK client.
       :param dbcluster_id: Cluster ID.
       :param resource_group_name: Name of the Spark Interactive resource group.
       :param query: SQL statements to run. Separate multiple statements with semicolons.
       :param limit: Maximum number of rows to return.
       :param runtime_config: Spark SQL runtime properties (e.g., output format, partitions).
       :param schema: Default database name. Defaults to "default".
       :return: Query ID for polling.
       """
       req = ExecuteSparkWarehouseBatchSQLRequest()
       req.dbcluster_id = dbcluster_id
       req.resource_group_name = resource_group_name
       req.execute_time_limit_in_seconds = 3600       # Timeout: 1 hour
       req.schema = schema
       req.query = query
       req.execute_result_limit = limit
       if runtime_config:
           req.runtime_config = json.dumps(runtime_config)
   
       resp: ExecuteSparkWarehouseBatchSQLResponse = client.execute_spark_warehouse_batch_sql(req)
       logger.info("Query submitted: {}", resp.body.data.query_id)
       return resp.body.data.query_id
   
   
   def get_query_state(client, query_id):
       """
       Poll the execution status of a submitted query.
   
       :return: Tuple of (query_state, response). query_state is one of:
                PENDING   — queued; the Spark Interactive resource group may be starting.
                SUBMITTED — accepted by the resource group.
                RUNNING   — executing.
                FINISHED  — completed successfully.
                FAILED    — execution failed.
                CANCELED  — canceled by the user.
       """
       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_queries(client, db_cluster, resource_group_name, page_num):
       """
       List queries previously run in the Spark Interactive resource group (paginated, 10 per page).
   
       :return: True if there are no more pages, False otherwise.
       """
       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)
       if resp.body.data.queries is None:
           return True
       for query in resp.body.data.queries:
           logger.info("Query ID: {}, State: {}", query.query_id, query.query_state)
       logger.info("Total on this page: {}", 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("Reading {}", obj.key)
               csv_content = oss_client.get_object(obj.key).read().decode("utf-8")
               for row in csv.DictReader(StringIO(csv_content)):
                   print(row)
   
   
   if __name__ == "__main__":
       logger.info("ADB Spark SQL demo")
   
       _ak = "<your-access-key-id>"          # AccessKey ID
       _sk = "<your-access-key-secret>"      # AccessKey secret
       _region = "<region-id>"               # Region ID, e.g., cn-shanghai
       _db = "<cluster-id>"                  # AnalyticDB for MySQL cluster ID
       _rg_name = "<resource-group-name>"    # Spark Interactive resource group name
   
       client_config = Config(
           access_key_id=_ak,
           access_key_secret=_sk,
           endpoint=f"adb.{_region}.aliyuncs.com",   # Use adb-vpc.<region>.aliyuncs.com for VPC
       )
       _client = Client(client_config)
   
       _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://<your-bucket>/sql_result",
           spark_sql_runtime_config=_spark_sql_runtime_config,
       )
   
       _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("Waiting for query {} to finish...", _query_id)
   
       # Poll until the query finishes, fails, or times out (10 minutes).
       current_ts = time.time()
       while True:
           query_state, resp = get_query_state(_client, _query_id)
           if query_state == "FINISHED":
               logger.info("Query finished successfully")
               break
           elif query_state == "FAILED":
               logger.error("Query failed: {}", resp.body.data)
               exit(1)
           elif query_state == "CANCELED":
               logger.error("Query was canceled")
               exit(1)
           else:
               time.sleep(2)
               if time.time() - current_ts > 600:
                   logger.error("Query timed out after 10 minutes")
                   _client.cancel_spark_warehouse_batch_sql(
                       CancelSparkWarehouseBatchSQLRequest(query_id=_query_id)
                   )
                   exit(1)
   
       # A query can contain multiple statements. Iterate over each statement's result.
       for stmt in resp.body.data.statements:
           logger.info("Statement {}: result at {}", stmt.statement_id, 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(_ak, _sk),
               f"oss-{_region}.aliyuncs.com",
               _bucket,
           )
           list_csv_files(oss_client, _dir)
   
       # List execution history with pagination.
       logger.info("Listing query history")
       page_num = 1
       no_more_pages = list_history_queries(_client, _db, _rg_name, page_num)
       while not no_more_pages:
           page_num += 1
           no_more_pages = list_history_queries(_client, _db, _rg_name, page_num)

   Replace the following placeholders: