AnalyticDB:Analyse interactive Spark SQL

Console

1. Connectez-vous à la console AnalyticDB for MySQL. Dans le coin supérieur gauche de la console, sélectionnez une région. Dans le volet de navigation de gauche, cliquez sur Clusters. Recherchez le cluster que vous souhaitez gérer et cliquez sur son ID.

2. Dans le volet de navigation de gauche, choisissez Job Development > SQL Development.

3. Sélectionnez le moteur Spark et le groupe de ressources Spark Interactive créé, puis exécutez l'instruction Spark SQL suivante :

   SHOW DATABASES;

SDK

Lorsque vous exécutez des instructions Spark SQL à l'aide d'un SDK, les résultats de la requête sont écrits sous forme de fichiers dans un bucket OSS spécifié. Vous pouvez ensuite interroger les données dans la console OSS ou télécharger les fichiers de résultats sur votre ordinateur. L'exemple suivant montre comment appeler le SDK en Python.

1. Exécutez la commande suivante pour installer le SDK.

   pip install alibabacloud-adb20211201

2. Exécutez les commandes suivantes pour installer les dépendances.

   pip install oss2
   pip install loguru

3. Connectez-vous au cluster et exécutez des instructions 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 = "|"):
       """
       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)
   

   Paramètres :

   • _ak : l'AccessKey ID de votre compte Alibaba Cloud ou d'un utilisateur RAM disposant des autorisations d'accès sur AnalyticDB for MySQL. Pour savoir comment obtenir un AccessKey ID et un AccessKey Secret, consultez la section Comptes et autorisations.

   • _sk : l'AccessKey Secret de votre compte Alibaba Cloud ou d'un utilisateur RAM disposant des autorisations d'accès sur AnalyticDB for MySQL. Pour savoir comment obtenir un AccessKey ID et un AccessKey Secret, consultez la section Comptes et autorisations.

   • _region : l'ID de la région où réside votre cluster AnalyticDB for MySQL.

   • _db : l'ID du cluster AnalyticDB for MySQL.

   • _rg_name : le nom du groupe de ressources Spark Interactive.

   • oss_location (facultatif) : le chemin OSS où sont stockés les fichiers de résultats de requête.

     Si vous ne spécifiez pas ce paramètre, vous ne pourrez consulter que les cinq premières lignes du résultat de la requête dans le journal Log de la page Job Development > Spark JAR Development.

Applications

Clients

Outre les clients Beeline, DBeaver, DBVisualizer et DataGrip décrits dans cette rubrique, vous pouvez également effectuer des analyses interactives à l'aide d'outils d'ordonnancement de workflows tels que Airflow, Azkaban et DolphinScheduler.