：Gunakan Spark untuk mengimpor data

Prasyarat

• Versi instance Hologres Anda adalah V0.9 atau lebih baru. Anda dapat melihat versi instance Hologres di halaman detail instance pada konsol Hologres. Jika versi instance Hologres Anda lebih lama dari V0.9, tingkatkan instance Hologres secara manual di konsol Hologres atau bergabunglah dengan grup DingTalk Hologres untuk mendapatkan dukungan teknis. Untuk informasi lebih lanjut tentang cara menaikkan versi instance Hologres secara manual di konsol Hologres, lihat Peningkatan Instance. Untuk informasi lebih lanjut tentang cara mendapatkan dukungan teknis, lihat Dapatkan Dukungan Online untuk Hologres.

• Lingkungan Spark dengan versi yang didukung oleh Hologres telah diinstal. Dengan cara ini, Anda dapat menjalankan perintah spark-shell di lingkungan Spark.

Penggunaan koneksi

Saat menggunakan konektor Spark Hologres untuk membaca atau menulis data, koneksi Java Database Connectivity (JDBC) digunakan. Jumlah koneksi yang digunakan dipengaruhi oleh item-item berikut:

• Jumlah tugas Spark paralel: Anda dapat memperoleh jumlah tersebut di UI Spark saat pekerjaan sedang berjalan.

• Mode penulisan data: Jika Anda menggunakan konektor Spark untuk menulis data dalam mode salinan tetap, setiap tugas paralel menggunakan satu koneksi JDBC. Jika Anda menggunakan pernyataan INSERT untuk menulis data, jumlah koneksi JDBC yang digunakan oleh setiap tugas paralel sama dengan nilai write_thread_size. Setiap tugas paralel untuk pembacaan data menggunakan satu koneksi JDBC.

• Lainnya: Saat pekerjaan dimulai, informasi skema mungkin perlu diperoleh. Dalam hal ini, koneksi JDBC mungkin digunakan untuk waktu yang singkat.

Jumlah total koneksi JDBC yang digunakan oleh pekerjaan dapat dihitung menggunakan rumus berikut:

• Mode salinan tetap: Paralelisme × 1 + 1

• Pernyataan INSERT: Paralelisme × write_thread_size + 1

(Direkomendasikan) Gunakan konektor Spark untuk menulis data ke Hologres

Kami merekomendasikan agar Anda menggunakan konektor Spark bawaan Hologres untuk menulis data ke Hologres. Konektor Spark digunakan bersama dengan Holo Client. Dibandingkan dengan metode lain untuk menulis data, konektor Spark memberikan performa penulisan yang lebih baik. Untuk menggunakan konektor Spark untuk menulis data, ikuti langkah-langkah berikut. Untuk informasi lebih lanjut tentang kode contoh, lihat bagian Contoh Penggunaan Konektor Spark untuk Menulis Data ke Hologres dalam topik ini.

Gunakan Spark untuk membaca data dari jenis sumber tertentu dan menulis data tersebut ke Hologres

1. Baca data dari jenis sumber tertentu.

   Anda dapat menggunakan Spark untuk membaca data dari berbagai jenis sumber. Contoh-contoh berikut menunjukkan cara membaca data dari Hologres atau jenis sumber lainnya:

   • Baca data dari Hologres

     Hologres kompatibel dengan PostgreSQL. Anda dapat menggunakan Spark untuk membaca data Hologres berdasarkan driver JDBC PostgreSQL. Contoh kode berikut hanya untuk referensi.

     Catatan

     Sebelum membaca data Hologres, unduh paket JAR PostgreSQL JDBC JAR di situs resmi. Dalam contoh ini, postgresql-42.2.18 digunakan. Kemudian, jalankan perintah ./bin/spark-shell --jars /path/to/postgresql-42.2.18.jar di spark-shell untuk memuat paket JAR JDBC PostgreSQL. Anda juga dapat memuat paket JAR JDBC PostgreSQL bersama dengan paket JAR konektor Spark Hologres.

     // Baca dari beberapa tabel, misalnya: tb008
     val readDf = spark.read
       .format("jdbc") // Baca data Hologres berdasarkan driver JDBC PostgreSQL.
       .option("driver","org.postgresql.Driver")
       .option("url", "jdbc:postgresql://Ip:Por/test_database")
       .option("dbtable", "tb008")
       .option("user", "your_username")
       .option("password", "your_password")
       .load()

     Konektor Spark V1.3.2 dan yang lebih baru memungkinkan Anda membaca data dari Hologres dan mengoptimalkan performa pembacaan data paralel. Dibandingkan dengan driver JDBC PostgreSQL, konektor Spark memungkinkan Anda mengonfigurasi paralelisme pembacaan data dan mendukung sharding data di Hologres untuk mengimplementasikan pembacaan data paralel. Hal ini secara signifikan meningkatkan performa. Contoh kode:

     val spark = SparkSession
     .builder
     .appName("ReadFromHologres")
     .master("local[*]")
     .getOrCreate()
     
     spark.sparkContext.setLogLevel("WARN")
     
     import spark.implicits._
     
     val schema = StructType(Array(
       StructField("id", LongType),
       StructField("counts", IntegerType),
       StructField("name", StringType, false),
       StructField("price", DecimalType(38, 12)),
       StructField("out_of_stock", BooleanType)
     ))
     
     val readDf = spark.read
     .format("hologres")
     .schema(schema) // Opsional. Jika Anda tidak menentukan skema, semua bidang di tabel Hologres akan dibaca secara default.
     .option("username", "your_username")
     .option("password", "your_password")
     .option("jdbcurl", "jdbc:postgresql://hologres_endpoint/test_db")
     .option("table", "tb008")
     .option("scan_parallelism", "10") // Paralelisme default untuk membaca data dari Hologres. Nilai maksimumnya adalah jumlah shard tabel Hologres.
     .load()

   • Baca data dari jenis sumber lain seperti file Parquet

     Anda dapat menggunakan Spark untuk membaca data dari jenis sumber lain, seperti file Parquet atau tabel Hive. Contoh berikut memberikan kode contoh:

     import org.apache.spark.{SparkConf, SparkContext}
     import org.apache.spark.sql.hive.HiveContext
     
     val sparkConf = new SparkConf()
     val sc = new SparkContext(sparkConf)
     val hiveContext = new HiveContext(sc)
     
     // Baca dari beberapa tabel, misalnya: phone
     val readDf = hiveContext.sql("select * from hive_database.phone")

2. Tulis data tersebut ke Hologres.

   import com.alibaba.hologres.spark2.sink.SourceProvider
   
   -- Tulis ke tabel hologres
   df.write
     .format("hologres")
     .option(SourceProvider.USERNAME, "your_username")
     .option(SourceProvider.PASSWORD, "your_password")
     .option(SourceProvider.ENDPOINT, "Ip:Port")
     .option(SourceProvider.DATABASE, "test_database")
     .option(SourceProvider.TABLE, table)
     .option(SourceProvider.WRITE_BATCH_SIZE, 512) -- Jumlah maksimum permintaan yang diizinkan dalam satu batch.
     .option(SourceProvider.INPUT_DATA_SCHEMA_DDL, df.schema.toDDL) -- Parameter ini hanya diperlukan untuk Spark 3.X.
     .mode(SaveMode.Append) // Parameter ini hanya diperlukan untuk Spark 3.X.
     .save()

Gunakan Spark untuk menulis data ke Hologres secara real-time

1. Eksekusi pernyataan berikut untuk membuat tabel tempat Anda ingin menulis data di Hologres:

   CREATE TABLE test_table_stream
   (
       value text,
       count bigint
   );

2. Baca data dari mesin lokal Anda. Kumpulkan statistik frekuensi kata dan tulis statistik tersebut ke Hologres secara real-time. Contoh berikut memberikan kode contoh:

   • Kode

      val spark = SparkSession
           .builder
           .appName("StreamToHologres")
           .master("local[*]")
           .getOrCreate()
     
         spark.sparkContext.setLogLevel("WARN")
         import spark.implicits._
     
         val lines = spark.readStream
           .format("socket")
           .option("host", "localhost")
           .option("port", 9999)
           .load()
     
         -- Pisahkan baris menjadi kata-kata
         val words = lines.as[String].flatMap(_.split(" "))
     
         -- Hasilkan hitungan kata berjalan
         val wordCounts = words.groupBy("value").count()
     
         wordCounts.writeStream
             .outputMode(OutputMode.Complete())
             .format("hologres")
             .option(SourceProvider.USERNAME, "your_username")
             .option(SourceProvider.PASSWORD, "your_password")
             .option(SourceProvider.JDBCURL, "jdbc:postgresql://Ip:Port/dbname")
             .option(SourceProvider.TABLE, "test_table_stream")
             .option("batchsize", 1)
             .option("isolationLevel", "NONE")
             .option("checkpointLocation", checkpointLocation)
             .start()
             .awaitTermination()

   • Parameter

     Parameter	Nilai default	Diperlukan	Deskripsi
     username	Tidak ada nilai default	Ya	ID AccessKey akun Alibaba Cloud Anda. Anda dapat memperoleh ID AccessKey di halaman Manajemen Keamanan. Kami merekomendasikan agar Anda mengonfigurasi variabel lingkungan dan memperoleh ID AccessKey dan Rahasia AccessKey dari variabel lingkungan. Ini membantu mengurangi risiko kebocoran.
     password	Tidak ada nilai default	Ya	Rahasia AccessKey akun Alibaba Cloud Anda. Anda dapat memperoleh Rahasia AccessKey di halaman Manajemen Keamanan. Kami merekomendasikan agar Anda mengonfigurasi variabel lingkungan dan memperoleh ID AccessKey dan Rahasia AccessKey dari variabel lingkungan. Ini membantu mengurangi risiko kebocoran.
     table	Tidak ada nilai default	Ya	Nama tabel Hologres tempat Anda ingin menulis data.
     endpoint	Tidak ada nilai default	Konfigurasikan parameter ini atau parameter JDBCURL.	Titik akhir instance Hologres Anda. Anda dapat memperoleh titik akhir instance Hologres Anda di bagian Network Information halaman detail instance di konsol Hologres.
     database	Tidak ada nilai default	Konfigurasikan parameter ini atau parameter JDBCURL.	Nama database Hologres tempat tabel tujuan berada.
     jdbcurl	Tidak ada nilai default	Konfigurasikan hanya parameter ini atau parameter ENDPOINT dan DATABASE.	JDBC URL dari instance Hologres Anda.
     copy_write_mode	true	Tidak	Menentukan apakah akan menulis data dalam mode salinan tetap. Salinan tetap adalah fitur baru yang didukung di Hologres V1.3. Dalam mode salinan tetap, data ditulis dalam mode streaming daripada dalam batch. Oleh karena itu, penulisan data dalam mode salinan tetap memberikan throughput lebih tinggi dan latensi data lebih rendah, serta mengonsumsi lebih sedikit sumber daya memori klien dibandingkan dengan penulisan data menggunakan pernyataan INSERT. Catatan Jika Anda ingin menggunakan mode salinan tetap, versi konektor harus V1.3.0 atau lebih baru, dan versi mesin Hologres harus V1.3.34 atau lebih baru.
     copy_write_format	false	Tidak	Menentukan apakah akan memeriksa data kotor. Parameter ini hanya berlaku saat COPY_WRITE_MODE disetel ke true. Jika Anda menyetel parameter ini ke true dan data kotor dihasilkan, sistem dapat menemukan baris yang gagal ditulis. Catatan Pemeriksaan data kotor berdampak negatif pada performa penulisan. Kami merekomendasikan agar Anda menyetel parameter ini ke true hanya selama proses pemecahan masalah.
     bulk_load	true	Tidak	Menentukan apakah akan menulis data ke Hologres dalam mode salinan batch. Dalam mode salinan tetap, data ditulis dalam mode streaming. Catatan Dalam Hologres V2.1, performa penulisan data ke tabel tanpa kunci utama dioptimalkan. Dalam Hologres V2.1, operasi penulisan data ke tabel tanpa kunci utama dalam mode batch tidak memperoleh kunci tabel tetapi memperoleh kunci baris. Dengan cara ini, penulisan data dalam mode batch dapat dilakukan secara paralel dengan operasi yang dilakukan menggunakan rencana tetap. Ini membantu meningkatkan efisiensi pemrosesan data dan paralelisme. Jika versi konektor adalah V1.4.0 atau lebih baru, versi mesin Hologres harus V2.1.0 atau lebih baru.
     max_cell_buffer_size	20971520 (20 MB)	Tidak	Panjang maksimum bidang saat COPY_WRITE_MODE disetel ke true.
     copy_write_dirty_data_check	false	Tidak	Menentukan apakah akan memeriksa data kotor. Jika Anda menyetel parameter ini ke true dan data kotor dihasilkan, sistem dapat menemukan baris yang gagal ditulis. Pemeriksaan data kotor berdampak negatif pada performa penulisan. Kami merekomendasikan agar Anda menyetel parameter ini ke true hanya selama proses pemecahan masalah. Catatan Parameter ini hanya berlaku saat COPY_WRITE_MODE disetel ke true.
     copy_write_direct_connect	true untuk skenario yang mengizinkan koneksi langsung	Tidak	Parameter ini hanya berlaku saat COPY_WRITE_MODE disetel ke true. Jumlah data yang dapat ditulis dalam mode salinan ditentukan berdasarkan throughput titik akhir VPC. Sistem memeriksa apakah lingkungan dapat terhubung langsung ke node FE Hologres saat data ditulis dalam mode salinan. Koneksi langsung digunakan secara default jika lingkungan dapat terhubung langsung ke node FE Hologres. Jika parameter ini disetel ke false, koneksi langsung tidak digunakan.
     input_data_schema_ddl	Tidak ada nilai default	Parameter ini diperlukan untuk Spark 3.X. Konfigurasikan nilai parameter dalam format <your_DataFrame>.schema.toDDL.	Pernyataan DDL untuk DataFrame di Spark.
     write_mode	INSERT_OR_REPLACE	Tidak	Kebijakan yang digunakan untuk menangani konflik kunci utama. Parameter ini diperlukan jika tabel tujuan memiliki kunci utama. Nilai valid: INSERT_OR_IGNORE: membuang data yang ingin Anda tulis jika terjadi konflik kunci utama. INSERT_OR_UPDATE: memperbarui kolom relevan di tabel tujuan jika terjadi konflik kunci utama. INSERT_OR_REPLACE: memperbarui semua kolom di tabel tujuan jika terjadi konflik kunci utama.
     write_batch_size	512	Tidak	Jumlah maksimum permintaan yang diizinkan dalam satu batch dalam thread untuk menulis data. Jika jumlah total permintaan PUT mencapai batas atas yang ditentukan oleh parameter write_batch_size setelah konflik ditangani berdasarkan parameter write_mode, data dikirimkan dalam batch.
     write_batch_byte_size	2 MB	Tidak	Jumlah maksimum byte yang diizinkan dalam satu batch dalam thread untuk menulis data. Jika jumlah total byte dari permintaan PUT mencapai batas atas yang ditentukan oleh parameter WRITE_BATCH_BYTE_SIZE setelah konflik ditangani berdasarkan parameter WRITE_MODE, data dikirimkan dalam batch.
     write_max_interval_ms	10000 ms	Tidak	Interval pengiriman data dalam batch.
     write_fail_strategy	TYR_ONE_BY_ONE	Tidak	Kebijakan yang digunakan untuk menangani kegagalan pengiriman. Jika batch tidak dapat dikirimkan, Holo Client mengirimkan entri data satu per satu dalam batch sesuai urutan tertentu. Jika entri data tidak dapat dikirimkan, Holo Client mengembalikan pesan kesalahan yang berisi informasi tentang entri data tersebut.
     write_thread_size	1	Tidak	Jumlah thread paralel yang digunakan untuk menulis data. Setiap thread menempati satu koneksi. Jumlah total koneksi yang ditempati oleh pekerjaan Spark bervariasi berdasarkan paralelisme Spark. Jumlah total koneksi dapat dihitung menggunakan rumus berikut: Jumlah total koneksi = Nilai spark.default.parallelism × Nilai WRITE_THREAD_SIZE.
     dynamic_partition	false	Tidak	Menentukan apakah akan membuat partisi secara otomatis jika data ditulis ke tabel induk yang tidak memiliki partisi. Nilai valid: true dan false. true: Partisi dibuat secara otomatis jika data ditulis ke tabel induk yang tidak memiliki partisi.
     retry_count	3	Tidak	Jumlah maksimum percobaan ulang yang diizinkan untuk menulis dan mengkueri data jika terjadi kegagalan koneksi.
     retry_sleep_init_ms	1000 ms	Tidak	Jumlah waktu yang dikonsumsi oleh percobaan ulang untuk permintaan dihitung menggunakan rumus berikut: retry_sleep_init_ms + retry_count × retry_sleep_step_ms.
     retry_sleep_step_ms	10*1000 ms	Tidak	Jumlah waktu yang dikonsumsi oleh percobaan ulang untuk permintaan dihitung menggunakan rumus berikut: retry_sleep_init_ms + retry_count × retry_sleep_step_ms.
     connection_max_idle_ms	60000 ms	Tidak	Periode timeout idle untuk koneksi yang digunakan untuk membaca dan menulis data. Jika koneksi tetap idle selama periode waktu lebih lama dari periode timeout idle yang ditentukan, Holo Client secara otomatis melepaskan koneksi.
     fixed_connection_mode	false	Tidak	Menentukan apakah akan menggunakan koneksi tetap. Dalam mode salinan non-tetap, seperti mode INSERT, penulisan data dan kueri titik tidak menempati koneksi. Catatan Fitur koneksi tetap masih dalam versi beta dan hanya tersedia saat versi konektor adalah V1.2.0 atau lebih baru dan versi mesin Hologres adalah V1.3.0 atau lebih baru.
     scan_batch_size	256	Tidak	Jumlah baris yang dipindai setiap kali saat Anda membaca data dari Hologres.
     scan_timeout_seconds	60	Tidak	Periode timeout operasi pemindaian saat Anda membaca data dari Hologres. Unit: detik.
     scan_parallelism	10	Tidak	Jumlah shard yang digunakan untuk membaca data dari Hologres. Nilai maksimumnya adalah jumlah shard tabel Hologres. Selama runtime pekerjaan, shard dialokasikan ke tugas Spark untuk pembacaan data.

Pemetaan tipe data

Tabel berikut menjelaskan pemetaan tipe data antara Spark dan Hologres.