AnalyticDB:Spark SQL を使用した対話型分析

コンソール

1. AnalyticDB for MySQL コンソールにログインします。コンソールの左上隅でリージョンを選択します。左側のナビゲーションウィンドウで、クラスターリスト をクリックします。管理するクラスターを見つけて、クラスター ID をクリックします。

2. 左側のナビゲーションウィンドウで、[ジョブ開発] > [SQL 開発] を選択します。

3. Spark エンジンと作成した Spark 対話型リソースグループを選択します。次に、次の Spark SQL 文を実行します。

   SHOW DATABASES;

SDK 呼び出し

SDK メソッドを呼び出して Spark SQL クエリを実行すると、クエリ結果は指定された OSS の場所にファイルとして書き込まれます。その後、OSS コンソールでデータをクエリするか、「結果ファイルをローカルコンピューターにダウンロード」して表示できます。次のセクションでは、Python を使用して SDK を呼び出す例を示します。

1. 次のコマンドを実行して SDK をインストールします。

   pip install alibabacloud-adb20211201

2. 次のコマンドを実行して依存関係をインストールします。

   pip install oss2
   pip install loguru

3. クラスターに接続し、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 = "|"):
       """
       ADB SQL 実行の構成を構築します。
       :param oss_location: SQL 実行結果が格納される OSS パス。
       :param spark_sql_runtime_config: Spark SQL コミュニティのネイティブ構成。
       :param file_format: SQL 実行結果のファイル形式。デフォルト値: CSV。
       :param output_partitions: SQL 実行結果のパーティション数。大量の結果を出力する必要がある場合は、単一のファイルが大きくなりすぎないように、出力パーティションの数を増やす必要があります。
       :param sep: CSV ファイルの区切り文字。このパラメーターは CSV 以外のファイルでは無視されます。
       :return: SQL 実行の構成。
       """
   
       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"
                   ):
       """
       Spark 対話型リソースグループで SQL 文を実行します。
       :param client: Alibaba Cloud Client。
       :param dbcluster_id: クラスターの ID。
       :param resource_group_name: クラスターのリソースグループ。Spark 対話型リソースグループである必要があります。
       :param schema: SQL 実行のデフォルトのデータベース名。指定しない場合、デフォルト値は `default` です。
       :param limit: SQL 実行結果の最大行数。
       :param query: 実行する SQL 文。セミコロン (;) を使用して複数の SQL 文を区切ります。
       :return:
       """
       # リクエストボディを組み立てます。
       req = ExecuteSparkWarehouseBatchSQLRequest()
       # クラスター ID。
       req.dbcluster_id = dbcluster_id
       # リソースグループ名。
       req.resource_group_name = resource_group_name
       # SQL 実行のタイムアウト期間。
       req.execute_time_limit_in_seconds = 3600
       # SQL 文が実行されるデータベースの名前。
       req.schema = schema
       # SQL ビジネスコード。
       req.query = query
       # 返される結果の行数。
       req.execute_result_limit = limit
   
       if runtime_config:
       # SQL 実行の構成。
           req.runtime_config = json.dumps(runtime_config) 
       # SQL 文を実行し、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):
       """
       SQL 文の実行ステータスをクエリします。
       :param client: Alibaba Cloud Client。
       :param query_id: SQL 実行の ID。
       :return: SQL 文の実行ステータスと結果。
       """
       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):
       """
       Spark 対話型リソースグループで実行された SQL 文の履歴をクエリします。
       :param client: Alibaba Cloud Client。
       :param db_cluster: クラスターの ID。
       :param resource_group_name: リソースグループ名。
       :param page_num: ページングクエリのページ番号。
       :return: SQL 文が存在するかどうか。存在する場合、次のページに進むことができます。
       """
       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)
   
       # SQL 文が見つからない場合は true を返します。それ以外の場合は false を返します。デフォルトのページサイズは 10 です。
       if resp.body.data.queries is None:
           return True
   
       # クエリされた SQL 文を出力します。
       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}")
               # oss ファイルの内容を読み取ります
               csv_content = oss_client.get_object(obj.key).read().decode('utf-8')
               csv_reader = csv.DictReader(StringIO(csv_content))
               # CSV の内容を出力します
               for row in csv_reader:
                   print(row)
   
   
   if __name__ == '__main__':
       logger.info("ADB Spark Batch SQL Demo")
   
       # AccessKey ID。 ご自身の AccessKey ID に置き換えてください。
       _ak = "LTAI****************"
       # AccessKey Secret。 ご自身の AccessKey Secret に置き換えてください。
       _sk = "yourAccessKeySecret"
       # リージョン ID。 ご自身のリージョン ID に置き換えてください。
       _region= "cn-shanghai" 
   
   
       # クラスター ID。 ご自身のクラスター ID に置き換えてください。
       _db = "amv-uf6485635f****"
       # リソースグループ名。 ご自身のリソースグループ名に置き換えてください。
       _rg_name = "testjob" 
   
       # クライアント構成
       client_config = Config(
           # Alibaba Cloud AccessKey ID
           access_key_id=_ak,
           # Alibaba Cloud AccessKey Secret
           access_key_secret=_sk,
           # ADB サービスのエンドポイント
           # adb.ap-southeast-1.aliyuncs.com はシンガポールリージョンの ADB サービスのエンドポイントです
           # adb-vpc.ap-southeast-1.aliyuncs.com は VPC シナリオで使用されます
           endpoint=f"adb.{_region}.aliyuncs.com"
       )
   
       # Alibaba Cloud Client を作成します。
       _client = Client(client_config)
   
       # SQL 実行の構成。
       _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)
       # 実行する SQL 文。
       _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...")
   
       # SQL 実行が完了するのを待ちます。
       current_ts = time.time()
       while True:
           query_state, resp = get_query_state(_client, _query_id)
           """
           query_state は次のいずれかの状態になります:
           - PENDING: ジョブはサービスキューで保留中です。Spark 対話型リソースグループが起動しています。
           - SUBMITTED: ジョブは Spark 対話型リソースグループに送信されました。
           - RUNNING: SQL 文が実行中です。
           - FINISHED: SQL 文はエラーなしで正常に実行されました。
           - FAILED: SQL 実行が失敗しました。
           - CANCELED: SQL 実行がキャンセルされました。
           """
   
           if query_state == "FINISHED":
               logger.info("query finished success")
               break
           elif query_state == "FAILED":
               # 失敗情報を出力します。
               logger.error("Error Info: {}", resp.body.data)
               exit(1)
           elif query_state == "CANCELED":
               # キャンセル情報を出力します。
               logger.error("query canceled")
               exit(1)
           else:
               time.sleep(2)
               if time.time() - current_ts > 600:
                   logger.error("query timeout")
                   # 実行時間が 10 分を超えた場合、SQL 実行をキャンセルします。
                   _client.cancel_spark_warehouse_batch_sql(CancelSparkWarehouseBatchSQLRequest(query_id=_query_id))
                   exit(1)
   
       # 1 つのクエリに複数の文を含めることができます。すべての文をリストします。
       for stmt in resp.body.data.statements:
           logger.info(
               f"statement_id: {stmt.statement_id}, result location: {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(client_config.access_key_id, client_config.access_key_secret),
                                    f"oss-{_region}.aliyuncs.com", _bucket)
           list_csv_files(oss_client, _dir)
   
       # Spark 対話型リソースグループで実行されたすべての SQL 文をクエリします。ページングクエリがサポートされています。
       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)
   

   パラメーター:

   • _ak: AnalyticDB for MySQL の アクセス権限 を持つ Alibaba Cloud アカウントまたは RAM ユーザーの AccessKey ID。AccessKey ID と AccessKey Secret の取得方法の詳細については、「アカウントと権限」をご参照ください。

   • _sk: AnalyticDB for MySQL の アクセス権限 を持つ Alibaba Cloud アカウントまたは RAM ユーザーの AccessKey Secret。AccessKey ID と AccessKey Secret の取得方法の詳細については、「アカウントと権限」をご参照ください。

   • region: AnalyticDB for MySQL クラスターのリージョン ID。

   • _db: AnalyticDB for MySQL クラスターの ID。

   • _rg_name: Spark 対話型リソースグループの名前。

   • oss_location: (オプション) クエリ結果ファイルが格納される OSS パス。

     このパラメーターを指定しない場合、[ジョブ開発] > [Spark Jar 開発] ページの SQL クエリ文の [ログ] で、データの最初の 5 行のみを表示できます。

アプリケーション

クライアント

このトピックで説明する Beeline、DBeaver、DBVisualizer、Datagrip クライアントに加えて、「Airflow」、「Azkaban」、「DolphinScheduler」などのワークフロースケジューリングツールでも対話型分析を実行できます。