Spark SQL interactive analysis

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 created Spark Interactive resource group, and then run the following Spark SQL statement:

   SHOW DATABASES;

SDK

When you run Spark SQL statements using an SDK, the query results are written as files to a specified OSS bucket. You can then query the data in the OSS console or download the result files to your computer. The following example shows how to call the SDK in 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 run Spark SQL statements.

   # 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 = "|"):
       """
       Builds the configuration for an AnalyticDB for MySQL SQL execution.
       :param oss_location: The OSS path to store the SQL execution results.
       :param spark_sql_runtime_config: The native Spark SQL configuration properties.
       :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 you need to output a large result set, you must increase this value to avoid creating a single oversized file.
       :param sep: The separator for CSV files. This parameter is ignored for non-CSV files.
       :return: The configuration for the 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"
                   ):
       """
       Runs 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. This must be a Spark Interactive resource group.
       :param schema: The name of the default database for the SQL execution. If you do not specify this parameter, the default value is used.
       :param limit: The number of rows to return for the SQL execution result.
       :param query: The SQL statement to run. Use semicolons (;) to separate multiple SQL statements.
       :return:
       """
       # Assemble the request body.
       req = ExecuteSparkWarehouseBatchSQLRequest()
       # The cluster ID.
       req.dbcluster_id = dbcluster_id
       # The name of the resource group.
       req.resource_group_name = resource_group_name
       # The timeout period for the SQL execution.
       req.execute_time_limit_in_seconds = 3600
       # The name of the database where the SQL statement is run.
       req.schema = schema
       # The SQL query or statements.
       req.query = query
       # The number of result rows to return.
       req.execute_result_limit = limit
   
       if runtime_config:
       # The configuration for the SQL execution.
           req.runtime_config = json.dumps(runtime_config) 
       # Submits the SQL statement and returns 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):
       """
       Queries 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):
       """
       Queries the history of SQL statements run 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 name of the resource group.
       :param page_num: The page number for paginated queries.
       :return: Specifies whether more pages are available. If queries exist, 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 True. The default is 10 entries per page.
       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")
   
       # Replace with your AccessKey ID.
       _ak = "LTAI****************"
       # Replace with your AccessKey secret.
       _sk = "yourAccessKeySecret"
       # Replace with the actual region ID.
       _region= "cn-shanghai" 
   
       # Replace with your cluster ID.
       _db = "amv-uf6485635f****"
       # Replace with the name of your resource group.
       _rg_name = "testjob" 
   
       # client config
       client_config = Config(
           # Your Alibaba Cloud AccessKey ID.
           access_key_id=_ak,
           # Your Alibaba Cloud AccessKey secret.
           access_key_secret=_sk,
           # The endpoint of the AnalyticDB for MySQL service.
           # adb.ap-southeast-1.aliyuncs.com is the endpoint of the service in the China (Singapore) region.
           # adb-vpc.ap-southeast-1.aliyuncs.com is used in VPC scenarios.
           endpoint=f"adb.{_region}.aliyuncs.com"
       )
   
       # Create an Alibaba Cloud client.
       _client = Client(client_config)
   
       # The configuration for the 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 statement to run.
       _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)
           """
           The query_state can be one of the following:
           - PENDING: The query is queued, which can occur while the Spark Interactive resource group is starting.
           - SUBMITTED: The query is submitted to the Spark Interactive resource group.
           - RUNNING: The SQL statement is executing.
           - FINISHED: The SQL execution completed successfully.
           - 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. The following loop processes each statement.
       for stmt in resp.body.data.statements:
           logger.info(
               f"statement_id: {stmt.statement_id}, result location: {stmt.result_uri}")
           # Sample code to view 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 run in the Spark Interactive resource group. You can perform paginated queries.
       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 your Alibaba Cloud account or a RAM user that has access permissions on AnalyticDB for MySQL. For information about how to obtain an AccessKey ID and an AccessKey secret, see Accounts and permissions.

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

   • _region: The ID of the region where your AnalyticDB for MySQL cluster resides.

   • _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 the query result in the Log on the Job Development > Spark JAR Development page.

Applications

Clients

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