AnalyticDB:Analisis interaktif Spark SQL

Konsol

1. Masuk ke Konsol AnalyticDB for MySQL. Di pojok kiri atas konsol, pilih wilayah. Di panel navigasi sebelah kiri, klik Clusters. Temukan kluster yang ingin Anda kelola lalu klik ID kluster tersebut.

2. Di panel navigasi sebelah kiri, pilih Job Development > SQL Development.

3. Pilih engine Spark dan kelompok sumber daya Spark Interactive yang telah dibuat, lalu jalankan pernyataan Spark SQL berikut:

   SHOW DATABASES;

SDK

Saat Anda menjalankan pernyataan Spark SQL menggunakan SDK, hasil kueri akan ditulis sebagai file ke bucket OSS yang ditentukan. Anda kemudian dapat mengkueri data tersebut di konsol OSS atau mengunduh file hasil ke komputer Anda. Contoh berikut menunjukkan cara memanggil SDK dalam Python.

1. Jalankan perintah berikut untuk menginstal SDK.

   pip install alibabacloud-adb20211201

2. Jalankan perintah berikut untuk menginstal dependensi.

   pip install oss2
   pip install loguru

3. Hubungkan ke kluster dan jalankan pernyataan 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 = "|"):
       """
       Membangun konfigurasi untuk eksekusi SQL AnalyticDB for MySQL.
       :param oss_location: Path OSS untuk menyimpan hasil eksekusi SQL.
       :param spark_sql_runtime_config: Properti konfigurasi native Spark SQL.
       :param file_format: Format file hasil eksekusi SQL. Nilai default: CSV.
       :param output_partitions: Jumlah partisi untuk hasil eksekusi SQL. Jika Anda perlu mengeluarkan set hasil yang besar, tingkatkan nilai ini untuk menghindari pembuatan satu file yang terlalu besar.
       :param sep: Pemisah untuk file CSV. Parameter ini diabaikan untuk file non-CSV.
       :return: Konfigurasi untuk eksekusi 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"
                   ):
       """
       Menjalankan pernyataan SQL di kelompok sumber daya Spark Interactive.
       :param client: Klien Alibaba Cloud.
       :param dbcluster_id: ID kluster.
       :param resource_group_name: Kelompok sumber daya kluster. Ini harus merupakan kelompok sumber daya Spark Interactive.
       :param schema: Nama database default untuk eksekusi SQL. Jika Anda tidak menentukan parameter ini, nilai default akan digunakan.
       :param limit: Jumlah baris yang dikembalikan untuk hasil eksekusi SQL.
       :param query: Pernyataan SQL yang akan dijalankan. Gunakan titik koma (;) untuk memisahkan beberapa pernyataan SQL.
       :return:
       """
       # Merakit badan permintaan.
       req = ExecuteSparkWarehouseBatchSQLRequest()
       # ID kluster.
       req.dbcluster_id = dbcluster_id
       # Nama kelompok sumber daya.
       req.resource_group_name = resource_group_name
       # Batas waktu eksekusi SQL.
       req.execute_time_limit_in_seconds = 3600
       # Nama database tempat pernyataan SQL dijalankan.
       req.schema = schema
       # Kueri atau pernyataan SQL.
       req.query = query
       # Jumlah baris hasil yang dikembalikan.
       req.execute_result_limit = limit
   
       if runtime_config:
       # Konfigurasi untuk eksekusi SQL.
           req.runtime_config = json.dumps(runtime_config) 
       # Mengirimkan pernyataan SQL dan mengembalikan ID kueri.
       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):
       """
       Mengkueri status eksekusi pernyataan SQL.
       :param client: Klien Alibaba Cloud.
       :param query_id: ID eksekusi SQL.
       :return: Status eksekusi dan hasil pernyataan 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):
       """
       Mengkueri riwayat pernyataan SQL yang dijalankan di kelompok sumber daya Spark Interactive.
       :param client: Klien Alibaba Cloud.
       :param db_cluster: ID kluster.
       :param resource_group_name: Nama kelompok sumber daya.
       :param page_num: Nomor halaman untuk kueri terpaginasi.
       :return: Menentukan apakah masih ada halaman berikutnya. Jika kueri ada, Anda dapat melanjutkan ke halaman berikutnya.
       """
       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)
   
       # Jika tidak ditemukan pernyataan SQL, kembalikan True. Jika tidak, kembalikan True. Default adalah 10 entri per halaman.
       if resp.body.data.queries is None:
           return True
   
       # Mencetak pernyataan SQL yang dikueri.
       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}")
               # membaca konten file oss
               csv_content = oss_client.get_object(obj.key).read().decode('utf-8')
               csv_reader = csv.DictReader(StringIO(csv_content))
               # Mencetak konten CSV
               for row in csv_reader:
                   print(row)
   
   if __name__ == '__main__':
       logger.info("ADB Spark Batch SQL Demo")
   
       # Ganti dengan ID AccessKey Anda.
       _ak = "LTAI****************"
       # Ganti dengan rahasia AccessKey Anda.
       _sk = "yourAccessKeySecret"
       # Ganti dengan ID wilayah yang sebenarnya.
       _region= "cn-shanghai" 
   
       # Ganti dengan ID kluster Anda.
       _db = "amv-uf6485635f****"
       # Ganti dengan nama kelompok sumber daya Anda.
       _rg_name = "testjob" 
   
       # konfigurasi klien
       client_config = Config(
           # ID AccessKey Alibaba Cloud Anda.
           access_key_id=_ak,
           # Rahasia AccessKey Alibaba Cloud Anda.
           access_key_secret=_sk,
           # Titik akhir layanan AnalyticDB for MySQL.
           # adb.ap-southeast-1.aliyuncs.com adalah titik akhir layanan di wilayah Tiongkok (Singapura).
           # adb-vpc.ap-southeast-1.aliyuncs.com digunakan dalam skenario VPC.
           endpoint=f"adb.{_region}.aliyuncs.com"
       )
   
       # Membuat klien Alibaba Cloud.
       _client = Client(client_config)
   
       # Konfigurasi untuk eksekusi 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)
       # Pernyataan SQL yang akan dijalankan.
       _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...")
   
       # Menunggu eksekusi SQL selesai.
       current_ts = time.time()
       while True:
           query_state, resp = get_query_state(_client, _query_id)
           """
           query_state dapat berupa salah satu dari berikut:
           - PENDING: Kueri sedang dalam antrian, yang dapat terjadi saat kelompok sumber daya Spark Interactive sedang dimulai.
           - SUBMITTED: Kueri telah dikirim ke kelompok sumber daya Spark Interactive.
           - RUNNING: Pernyataan SQL sedang dieksekusi.
           - FINISHED: Eksekusi SQL berhasil selesai.
           - FAILED: Eksekusi SQL gagal.
           - CANCELED: Eksekusi SQL dibatalkan.
           """
   
           if query_state == "FINISHED":
               logger.info("query finished success")
               break
           elif query_state == "FAILED":
               # Mencetak informasi kegagalan.
               logger.error("Error Info: {}", resp.body.data)
               exit(1)
           elif query_state == "CANCELED":
               # Mencetak informasi pembatalan.
               logger.error("query canceled")
               exit(1)
           else:
               time.sleep(2)
               if time.time() - current_ts > 600:
                   logger.error("query timeout")
                   # Jika waktu eksekusi melebihi 10 menit, batalkan eksekusi SQL.
                   _client.cancel_spark_warehouse_batch_sql(CancelSparkWarehouseBatchSQLRequest(query_id=_query_id))
                   exit(1)
   
       # Satu kueri dapat berisi beberapa pernyataan. Loop berikut memproses setiap pernyataan.
       for stmt in resp.body.data.statements:
           logger.info(
               f"statement_id: {stmt.statement_id}, result location: {stmt.result_uri}")
           # Contoh kode untuk melihat hasil.
           _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)
   
       # Mengkueri semua pernyataan SQL yang dijalankan di kelompok sumber daya Spark Interactive. Anda dapat melakukan kueri terpaginasi.
       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)
   

   Parameter:

   • _ak: ID AccessKey akun Alibaba Cloud Anda atau pengguna RAM yang memiliki izin akses pada AnalyticDB for MySQL. Untuk informasi tentang cara mendapatkan ID AccessKey dan rahasia AccessKey, lihat Akun dan izin.

   • _sk: Rahasia AccessKey akun Alibaba Cloud Anda atau pengguna RAM yang memiliki izin akses pada AnalyticDB for MySQL. Untuk informasi tentang cara mendapatkan ID AccessKey dan rahasia AccessKey, lihat Akun dan izin.

   • _region: ID wilayah tempat kluster AnalyticDB for MySQL Anda berada.

   • _db: ID kluster AnalyticDB for MySQL.

   • _rg_name: Nama kelompok sumber daya Spark Interactive.

   • oss_location (opsional): Path OSS tempat file hasil kueri disimpan.

     Jika Anda tidak menentukan parameter ini, Anda hanya dapat melihat lima baris pertama hasil kueri di Log pada halaman Job Development > Spark JAR Development.

Aplikasi

Klien

Selain klien Beeline, DBeaver, DBVisualizer, dan DataGrip yang dijelaskan dalam topik ini, Anda juga dapat melakukan analisis interaktif di tool penjadwalan alur kerja seperti Airflow, Azkaban, dan DolphinScheduler.