AnalyticDB:Interactive analysis with Spark SQL

Console

1. Log on to the AnalyticDB for MySQL console. In the upper-left corner of the console, select a region. In the left-side navigation pane, click Clusters. Find the cluster that you want to manage and click the cluster ID.

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

3. Select the Spark engine and the Spark Interactive resource group that you created. Then, execute the following Spark SQL statement:

   SHOW DATABASES;

SDK call

When you call an SDK method to execute a Spark SQL query, the query result is written to a file in a specified OSS location. You can then query the data in the OSS console or download the result file to your local computer to view it. The following section provides an example of how to call the SDK using Python.

1. Run the following command to install the SDK.

   pip install alibabacloud-adb20211201

2. Run the following commands to install dependencies.

   pip install oss2
   pip install loguru

3. Connect to the cluster and execute Spark SQL queries.

   # 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 configuration for ADB SQL execution.
       :param oss_location: The OSS path where the SQL execution results are stored.
       :param spark_sql_runtime_config: The native configuration of the Spark SQL community.
       :param file_format: The file format of the SQL execution results. Default value: CSV.
       :param output_partitions: The number of partitions for the SQL execution results. If a large amount of results needs to be output, you must increase the number of output partitions to prevent a single file from being too large.
       :param sep: The separator for CSV files. This parameter is ignored for non-CSV files.
       :return: The configuration for SQL execution.
       """
   
       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"
                   ):
       """
       Execute an SQL statement in a Spark Interactive resource group.
       :param client: The Alibaba Cloud Client.
       :param dbcluster_id: The ID of the cluster.
       :param resource_group_name: The resource group of the cluster. It must be a Spark Interactive resource group.
       :param schema: The default database name for SQL execution. If not specified, the default value is `default`.
       :param limit: The maximum number of rows in the SQL execution result.
       :param query: The SQL statements to execute. Use semicolons (;) to separate multiple SQL statements.
       :return:
       """
       # Assemble the request body.
       req = ExecuteSparkWarehouseBatchSQLRequest()
       # The cluster ID.
       req.dbcluster_id = dbcluster_id
       # The resource group name.
       req.resource_group_name = resource_group_name
       # The timeout period for SQL execution.
       req.execute_time_limit_in_seconds = 3600
       # The name of the database where the SQL statement is executed.
       req.schema = schema
       # The SQL business code.
       req.query = query
       # The number of result rows to return.
       req.execute_result_limit = limit
   
       if runtime_config:
       # The configuration for SQL execution.
           req.runtime_config = json.dumps(runtime_config) 
       # Execute the SQL statement and obtain the 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):
       """
       Query the execution status of an SQL statement.
       :param client: The Alibaba Cloud Client.
       :param query_id: The ID of the SQL execution.
       :return: The execution status and result of the SQL statement.
       """
       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):
       """
       Query the history of SQL statements executed in a Spark Interactive resource group.
       :param client: The Alibaba Cloud Client.
       :param db_cluster: The ID of the cluster.
       :param resource_group_name: The resource group name.
       :param page_num: The page number for the paged query.
       :return: Whether an SQL statement exists. If so, you can proceed to the next page.
       """
       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 no SQL statement is found, return true. Otherwise, return false. The default page size is 10.
       if resp.body.data.queries is None:
           return True
   
       # Print the queried SQL statements.
       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")
   
       # The AccessKey ID. Replace the value with your AccessKey ID.
       _ak = "LTAI****************"
       # The AccessKey secret. Replace the value with your AccessKey secret.
       _sk = "yourAccessKeySecret"
       # The region ID. Replace the value with your region ID.
       _region= "cn-shanghai" 
   
   
       # The cluster ID. Replace the value with your cluster ID.
       _db = "amv-uf6485635f****"
       # The resource group name. Replace the value with your resource group name.
       _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"
       )
   
       # Create an Alibaba Cloud Client.
       _client = Client(client_config)
   
       # The configuration for SQL execution.
       _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)
       # The SQL statements to be executed.
       _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...")
   
       # Wait for the SQL execution to complete.
       current_ts = time.time()
       while True:
           query_state, resp = get_query_state(_client, _query_id)
           """
           query_state can be in one of the following states:
           - PENDING: The job is pending in the service queue. The Spark Interactive resource group is starting.
           - SUBMITTED: The job is submitted to the Spark Interactive resource group.
           - RUNNING: The SQL statement is being executed.
           - FINISHED: The SQL statement is executed successfully without errors.
           - FAILED: The SQL execution failed.
           - CANCELED: The SQL execution is canceled.
           """
   
           if query_state == "FINISHED":
               logger.info("query finished success")
               break
           elif query_state == "FAILED":
               # Print the failure information.
               logger.error("Error Info: {}", resp.body.data)
               exit(1)
           elif query_state == "CANCELED":
               # Print the cancellation information.
               logger.error("query canceled")
               exit(1)
           else:
               time.sleep(2)
               if time.time() - current_ts > 600:
                   logger.error("query timeout")
                   # If the execution time exceeds 10 minutes, cancel the SQL execution.
                   _client.cancel_spark_warehouse_batch_sql(CancelSparkWarehouseBatchSQLRequest(query_id=_query_id))
                   exit(1)
   
       # A query can contain multiple statements. List all statements.
       for stmt in resp.body.data.statements:
           logger.info(
               f"statement_id: {stmt.statement_id}, result location: {stmt.result_uri}")
           # Sample code for viewing the results.
           _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)
   
       # Query all SQL statements executed in the Spark Interactive resource group. Paged query is supported.
       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)
   

   Parameters:

   • _ak: The AccessKey ID of an Alibaba Cloud account or a RAM user that has AnalyticDB for MySQL access permissions. For more information about how to obtain an AccessKey ID and AccessKey secret, see Accounts and Permissions.

   • _sk: The AccessKey secret of an Alibaba Cloud account or a RAM user that has AnalyticDB for MySQL access permissions. For more information about how to obtain an AccessKey ID and AccessKey secret, see Accounts and Permissions.

   • region: The region ID of the AnalyticDB for MySQL cluster.

   • _db: The ID of the AnalyticDB for MySQL cluster.

   • _rg_name: The name of the Spark Interactive resource group.

   • oss_location: (Optional) The OSS path where the query result files are stored.

     If you do not specify this parameter, you can view only the first five rows of data in the Logs of the SQL query statement on the Job Development > Spark JAR Development page.

Applications

Clients

In addition to the Beeline, DBeaver, DBVisualizer, and Datagrip clients that are described in this topic, you can also perform interactive analysis in workflow scheduling tools such as Airflow, Azkaban, and DolphinScheduler.