AnalyticDB:Análise interativa com Spark SQL

Console

1. Faça login no console do AnalyticDB for MySQL. No canto superior esquerdo do console, selecione uma região. No painel de navegação à esquerda, clique em Clusters. Localize o cluster que deseja gerenciar e clique no ID do cluster.

2. No painel de navegação à esquerda, escolha Job Development > SQL Development.

3. Selecione o mecanismo Spark e o grupo de recursos Spark Interactive criado e execute a seguinte instrução Spark SQL:

   SHOW DATABASES;

SDK

Ao executar instruções Spark SQL usando um SDK, os resultados da consulta são gravados como arquivos em um bucket OSS especificado. Em seguida, consulte os dados no console do OSS ou baixe os arquivos de resultado para o seu computador. O exemplo a seguir mostra como chamar o SDK em Python.

1. Execute o comando a seguir para instalar o SDK.

   pip install alibabacloud-adb20211201

2. Execute os comandos a seguir para instalar as dependências.

   pip install oss2
   pip install loguru

3. Conecte-se ao cluster e execute instruções 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)
   

   Parâmetros:

   • _ak: O AccessKey ID da sua conta Alibaba Cloud ou de um usuário RAM com permissões de acesso ao AnalyticDB for MySQL. Para obter informações sobre como conseguir um AccessKey ID e um AccessKey secret, consulte Contas e permissões.

   • _sk: O AccessKey secret da sua conta Alibaba Cloud ou de um usuário RAM com permissões de acesso ao AnalyticDB for MySQL. Para obter informações sobre como conseguir um AccessKey ID e um AccessKey secret, consulte Contas e permissões.

   • _region: O ID da região onde reside o seu cluster AnalyticDB for MySQL.

   • _db: O ID do cluster AnalyticDB for MySQL.

   • _rg_name: O nome do grupo de recursos Spark Interactive.

   • oss_location (opcional): O caminho do OSS onde os arquivos de resultado da consulta são armazenados.

     Se você não especificar este parâmetro, poderá visualizar apenas as primeiras cinco linhas do resultado da consulta no Log na página Job Development > Spark JAR Development.

Aplicações

Clientes

Além dos clientes Beeline, DBeaver, DBVisualizer e DataGrip descritos neste tópico, você também pode realizar análises interativas em ferramentas de agendamento de fluxo de trabalho, como Airflow, Azkaban e DolphinScheduler.