AnalyticDB：Analisis interaktif dengan 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 dan 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 Anda buat. Kemudian, jalankan pernyataan Spark SQL berikut:

   SHOW DATABASES;

Panggilan SDK

Saat Anda memanggil metode SDK untuk menjalankan kueri Spark SQL, hasil kueri akan ditulis ke file di lokasi OSS yang ditentukan. Anda kemudian dapat mengkueri data tersebut di konsol OSS atau mengunduh file hasil ke komputer lokal Anda untuk melihatnya. Bagian berikut memberikan contoh cara memanggil SDK menggunakan 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 kueri 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 ADB.
       :param oss_location: Jalur OSS tempat hasil eksekusi SQL disimpan.
       :param spark_sql_runtime_config: Konfigurasi asli komunitas Spark SQL.
       :param file_format: Format file hasil eksekusi SQL. Nilai default: CSV.
       :param output_partitions: Jumlah partisi untuk hasil eksekusi SQL. Jika jumlah hasil yang besar perlu dikeluarkan, Anda harus menambah jumlah partisi output agar mencegah satu file 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 wajib diisi")
   
       if not oss_location.startswith("oss://"):
           raise ValueError("oss_location harus dimulai dengan oss://")
   
       if file_format != "CSV" and file_format != "PARQUET" and file_format != "ORC" and file_format != "JSON":
           raise ValueError("file_format harus berupa CSV, PARQUET, ORC, atau JSON")
   
       runtime_config =  {
           # konfigurasi output sql
           "spark.adb.sqlOutputFormat": file_format,
           "spark.adb.sqlOutputPartitions": output_partitions,
           "spark.adb.sqlOutputLocation": oss_location,
           # konfigurasi csv
           "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. Harus merupakan kelompok sumber daya Spark Interactive.
       :param schema: Nama database default untuk eksekusi SQL. Jika tidak ditentukan, nilai default adalah `default`.
       :param limit: Jumlah maksimum baris dalam 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
       # Kode bisnis 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) 
       # Jalankan pernyataan SQL dan dapatkan query_id.
       resp: ExecuteSparkWarehouseBatchSQLResponse = client.execute_spark_warehouse_batch_sql(req)
       logger.info("Permintaan eksekusi kueri dikirim: {}", resp.body.data.query_id)
       return resp.body.data.query_id
   
   
   def get_query_state(client, query_id):
       """
       Memeriksa status eksekusi pernyataan SQL.
       :param client: Klien Alibaba Cloud.
       :param query_id: ID eksekusi SQL.
       :return: Status dan hasil eksekusi SQL.
       """
       req = GetSparkWarehouseBatchSQLRequest(query_id=query_id)
       resp: GetSparkWarehouseBatchSQLResponse = client.get_spark_warehouse_batch_sql(req)
       logger.info("Status kueri: {}", resp.body.data.query_state)
       return resp.body.data.query_state, resp
   
   def list_history_query(client, db_cluster, resource_group_name, page_num):
       """
       Memeriksa riwayat pernyataan SQL yang dieksekusi 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 berhalaman.
       :return: Apakah ada pernyataan SQL. Jika ya, 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 false. Ukuran halaman default adalah 10.
       if resp.body.data.queries is None:
           return True
   
       # Cetak pernyataan SQL yang diperoleh.
       for query in resp.body.data.queries:
           logger.info("ID Kueri: {}, Status: {}", query.query_id, query.query_state)
       logger.info("Total kueri: {}", 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"membaca {obj.key}")
               # baca konten file oss
               csv_content = oss_client.get_object(obj.key).read().decode('utf-8')
               csv_reader = csv.DictReader(StringIO(csv_content))
               # Cetak konten CSV
               for row in csv_reader:
                   print(row)
   
   
   if __name__ == '__main__':
       logger.info("Demo ADB Spark Batch SQL")
   
       # ID AccessKey. Ganti nilainya dengan ID AccessKey Anda.
       _ak = "LTAI****************"
       # Rahasia AccessKey. Ganti nilainya dengan rahasia AccessKey Anda.
       _sk = "yourAccessKeySecret"
       # ID wilayah. Ganti nilainya dengan ID wilayah Anda.
       _region= "cn-shanghai" 
   
   
       # ID kluster. Ganti nilainya dengan ID kluster Anda.
       _db = "amv-uf6485635f****"
       # Nama kelompok sumber daya. Ganti nilainya dengan nama kelompok sumber daya Anda.
       _rg_name = "testjob" 
   
       # konfigurasi klien
       client_config = Config(
           # ID AccessKey Alibaba Cloud
           access_key_id=_ak,
           # Rahasia AccessKey Alibaba Cloud
           access_key_secret=_sk,
           # Titik akhir layanan ADB
           # adb.ap-southeast-1.aliyuncs.com adalah titik akhir layanan ADB di wilayah Singapura
           # adb-vpc.ap-southeast-1.aliyuncs.com digunakan dalam skenario VPC
           endpoint=f"adb.{_region}.aliyuncs.com"
       )
   
       # Buat 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"Menjalankan query_id: {_query_id} untuk SQL {_query}.\n Menunggu hasil...")
   
       # Tunggu hingga eksekusi SQL selesai.
       current_ts = time.time()
       while True:
           query_state, resp = get_query_state(_client, _query_id)
           """
           query_state dapat berada dalam salah satu status berikut:
           - PENDING: Pekerjaan sedang menunggu di antrean layanan. Kelompok sumber daya Spark Interactive sedang dimulai.
           - SUBMITTED: Pekerjaan telah dikirim ke kelompok sumber daya Spark Interactive.
           - RUNNING: Pernyataan SQL sedang dijalankan.
           - FINISHED: Pernyataan SQL berhasil dijalankan tanpa kesalahan.
           - FAILED: Eksekusi SQL gagal.
           - CANCELED: Eksekusi SQL dibatalkan.
           """
   
           if query_state == "FINISHED":
               logger.info("kueri berhasil selesai")
               break
           elif query_state == "FAILED":
               # Cetak informasi kegagalan.
               logger.error("Info Kesalahan: {}", resp.body.data)
               exit(1)
           elif query_state == "CANCELED":
               # Cetak informasi pembatalan.
               logger.error("kueri dibatalkan")
               exit(1)
           else:
               time.sleep(2)
               if time.time() - current_ts > 600:
                   logger.error("kueri melebihi batas waktu")
                   # 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. Cantumkan semua pernyataan.
       for stmt in resp.body.data.statements:
           logger.info(
               f"statement_id: {stmt.statement_id}, lokasi hasil: {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)
   
       # Periksa semua pernyataan SQL yang dieksekusi di kelompok sumber daya Spark Interactive. Kueri berhalaman didukung.
       logger.info("Menampilkan semua riwayat kueri")
       page_num = 1
       no_more_page = list_history_query(_client, _db, _rg_name, page_num)
       while no_more_page:
           logger.info(f"Menampilkan halaman {page_num}")
           page_num += 1
           no_more_page = list_history_query(_client, _db, _rg_name, page_num)
   

   Parameter:

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

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

   • region: ID wilayah kluster AnalyticDB for MySQL.

   • _db: ID kluster AnalyticDB for MySQL.

   • _rg_name: Nama kelompok sumber daya Spark Interactive.

   • oss_location: (Opsional) Jalur OSS tempat file hasil kueri disimpan.

     Jika Anda tidak menentukan parameter ini, Anda hanya dapat melihat lima baris data pertama di Log pernyataan kueri SQL 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 alat penjadwalan alur kerja seperti Airflow, Azkaban, dan DolphinScheduler.