E-MapReduce:Baca dari dan tulis ke Hologres

Contoh 1: Session SQL

Contoh ini menunjukkan cara membaca dari dan menulis ke Hologres menggunakan session SQL.

1. Buat session SQL. Untuk detailnya, lihat Kelola Session SQL.

   Saat membuat session, pilih koneksi jaringan yang telah Anda buat pada langkah sebelumnya dari daftar koneksi jaringan. Di bagian Spark Configuration, tambahkan parameter berikut untuk memuat hologres-connector-spark.

   # Tambahkan file JAR hologres-connector (hanya diperlukan untuk versi sebelum esr-4.8.0).
   spark.emr.serverless.user.defined.jars oss://<bucket>/hologres-connector-spark-3.x-<version>.jar
   
   # Konfigurasikan katalog Hologres.
   spark.sql.catalog.hologres_external_test_db com.alibaba.hologres.spark3.HoloTableCatalog
   spark.sql.catalog.hologres_external_test_db.username ***
   spark.sql.catalog.hologres_external_test_db.password ***
   spark.sql.catalog.hologres_external_test_db.jdbcurl jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/testdb

   Tabel berikut menjelaskan parameter-parameter tersebut.

   Parameter	Contoh	Deskripsi
   spark.emr.serverless.user.defined.jars	oss://<bucket>/hologres-connector-spark-3.x-<version>.jar	Menentukan path ke file JAR yang ditentukan pengguna.
   spark.sql.catalog.hologres_external_test_db	com.alibaba.hologres.spark3.HoloTableCatalog	Di Spark 3.x, parameter ini digunakan untuk mengonfigurasi sumber data Hologres sebagai katalog eksternal. Ini adalah nilai tetap.
   spark.sql.catalog.hologres_external_test_db.username	LTAI******	ID AccessKey Akun Alibaba Cloud Anda. Kami menyarankan Anda menggunakan secrets management untuk menangani informasi sensitif. Untuk detailnya, lihat Kelola informasi sensitif menggunakan secrets management.
   spark.sql.catalog.hologres_external_test_db.password	mXYV******	Rahasia AccessKey Akun Alibaba Cloud Anda. Kami menyarankan Anda menggunakan secrets management untuk menangani informasi sensitif. Untuk detailnya, lihat Kelola informasi sensitif menggunakan secrets management.
   spark.sql.catalog.hologres_external_test_db.jdbcurl	jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/testdb	URL koneksi JDBC instans Hologres.

   Anda dapat menyesuaikan bagian hologres_external_test_db dalam nama parameter.

2. Di halaman Data Development, buat job SparkSQL dan pilih session SQL yang telah Anda buat dari pojok kanan atas.

   Untuk informasi selengkapnya, lihat Pengembangan SparkSQL.

3. Salin kode berikut ke tab SparkSQL baru dan klik Run.

   -- Beralih ke database testdb.
   USE hologres_external_test_db;
   -- Tulis data.
   INSERT INTO `public`.test VALUES ('1004','tom');
   -- Kueri data.
   SELECT * FROM `public`.test;

   

Contoh 2: Job streaming

Contoh PySpark ini menunjukkan cara membaca data dari Kafka dan menuliskannya ke Hologres sebagai job streaming.

1. Dalam contoh kode ini, ganti informasi Kafka dan tabel Hologres dengan nilai aktual Anda.

   from pyspark.sql import SparkSession
   from pyspark.sql.functions import col
   
   # Konfigurasikan informasi Kafka Anda.
   servers = "alikafka-serverless-cn-xxxxx-vpc.alikafka.aliyuncs.com:9092"  # Ganti dengan bootstrap servers Kafka Anda.
   topic = "topic-name"  # Ganti dengan topik Kafka Anda.
   
   # Buat SparkSession.
   spark = SparkSession.builder \
       .appName("test read kafka") \
       .getOrCreate()
   
   # Baca stream Kafka.
   df = spark \
       .readStream \
       .format("kafka") \
       .option("kafka.bootstrap.servers", servers) \
       .option("subscribe", topic) \
       .load()
   
   # Definisikan fungsi untuk menulis ke Hologres (dipanggil untuk setiap micro-batch).
   def write_to_hologres(batch_df, batch_id):
       print(f"Writing batch {batch_id} to Hologres...")
       batch_df.write \
           .format("hologres") \
           .mode("append") \
           .insertInto("hologres_external_test_db.public.test")  # Ganti dengan tabel Hologres Anda.
   
   # Konversi key dan value ke string dan tulis stream-nya.
   query = df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)") \
       .writeStream \
       .foreachBatch(write_to_hologres) \
       .outputMode("append") \
       .trigger(processingTime='30 seconds') \
       .start()
   
   # Tunggu hingga kueri streaming berhenti (ini memblokir eksekusi cell di notebook).
   query.awaitTermination()

2. Unggah file.

   1. Di halaman Artifacts, klik Upload File.

   2. Di kotak dialog Upload File, klik area unggah untuk memilih file Python dari langkah sebelumnya, atau seret file ke area unggah.

3. Buat dan jalankan job streaming.

   1. Di halaman Development, klik ikon (Create).

   2. Di kotak dialog yang muncul, masukkan Name, pilih PySpark dari daftar Application (Streaming), lalu klik OK.

   3. Di tab pengembangan baru, konfigurasikan parameter berikut dan biarkan sisanya pada pengaturan default. Lalu, klik Publish.

      Parameter	Deskripsi
      Main Python Resources	Pilih file Python yang telah Anda unggah pada langkah sebelumnya.
      Engine Version	Pilih versi Spark yang kompatibel. Contoh ini menggunakan esr-4.6.0.
      Network Connection	Pilih koneksi jaringan yang telah Anda buat di Langkah 2.
      Spark Configuration	# Tambahkan file JAR hologres-connector (hanya diperlukan untuk versi sebelum esr-4.8.0). spark.emr.serverless.user.defined.jars oss://<bucket>/test_script/hologres-connector-spark-3.x-1.5.6-jar-with-dependencies.jar # Konfigurasikan katalog Hologres. spark.sql.catalog.hologres_external_test_db com.alibaba.hologres.spark3.HoloTableCatalog spark.sql.catalog.hologres_external_test_db.username *** spark.sql.catalog.hologres_external_test_db.password *** spark.sql.catalog.hologres_external_test_db.jdbcurl jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/testdb Untuk deskripsi detail parameter, lihat Contoh 1: Session SQL.

   4. Setelah Anda memublikasikan job, klik Go to O&M. Di halaman yang terbuka, klik Start.

4. Verifikasi hasilnya.

   1. Kirim pesan ke Kafka.

      

   2. Kueri data menggunakan Spark SQL.

Contoh 3: Session Notebook

1. Buat session Notebook. Untuk detailnya, lihat Kelola Session Notebook.

   Saat membuat session, pilih koneksi jaringan yang telah Anda buat pada langkah sebelumnya dari daftar koneksi jaringan. Di bagian Spark Configuration, tambahkan parameter berikut untuk memuat hologres-connector-spark.

   # Tambahkan file JAR hologres-connector (hanya diperlukan untuk versi sebelum esr-4.8.0).
   spark.emr.serverless.user.defined.jars oss://<bucket>/hologres-connector-spark-3.x-<version>.jar

2. Di halaman Data Development, buat job Notebook dan pilih session Notebook yang telah Anda buat dari pojok kanan atas.

3. Salin kode berikut ke tab Notebook baru dan klik .

   import pandas as pd
   from pyspark.sql import SparkSession
   from pyspark.sql.types import StructType, StructField, StringType, IntegerType, LongType
   
   # 1. Siapkan DataFrame Pandas.
   pdf = pd.DataFrame({
       "id": ["1006"],
       "name": ["sl"]
   })
   
   # 2. Konversi ke DataFrame PySpark.
   # (Opsional: Definisikan skema secara eksplisit untuk memastikan tipe data yang benar)
   schema = StructType([
       StructField("id", StringType(), True),
       StructField("name", StringType(), True)
   ])
   
   df = spark.createDataFrame(pdf, schema=schema)
   
   # Tulis ke Hologres.
   df.write \
     .format("hologres") \
     .option("username", "LTAI******") \
     .option("password", "mXYV******") \
     .option("jdbcurl", "jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/testdb") \
     .option("table", "test") \
     .mode("append") \
     .save()
   
   # Baca data.
   readDf = spark.read\
     .format("hologres") \
     .option("username", "LTAI******") \
     .option("password", "mXYV******") \
     .option("jdbcurl", "jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/testdb") \
     .option("table", "test") \
     .load()
   
   readDf.select("id", "name").show(10)

   Tabel berikut menjelaskan parameter-parameter tersebut.

   Parameter	Contoh	Deskripsi
   username	LTAI******	ID AccessKey Akun Alibaba Cloud Anda. Kami menyarankan Anda menggunakan secrets management untuk menangani informasi sensitif. Untuk detailnya, lihat Kelola informasi sensitif menggunakan secrets management.
   password	mXYV******	Rahasia AccessKey Akun Alibaba Cloud Anda. Kami menyarankan Anda menggunakan secrets management untuk menangani informasi sensitif. Untuk detailnya, lihat Kelola informasi sensitif menggunakan secrets management.
   jdbcurl	jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/testdb	URL koneksi JDBC instans Hologres.

4. Verifikasi hasilnya.