Baca dan tulis Hologres dengan Spark - Hologres - Alibaba Cloud Documentation Center

Spark adalah mesin analitik terpadu untuk pemrosesan data skala besar. Hologres terintegrasi secara efisien dengan Spark komunitas dan EMR Serverless Spark guna membangun gudang data dengan cepat. Konektor Spark Hologres mendukung pembuatan katalog Hologres dalam kluster Spark, memungkinkan pembacaan batch berkinerja tinggi dan impor melalui tabel eksternal yang mengungguli JDBC asli.

Batasan

Konektor Spark memerlukan Hologres versi 1.3 atau lebih baru. Anda dapat memeriksa versi instans Anda di halaman Instance Details di Hologres console. Jika instans Anda menggunakan versi sebelum 1.3, upgrade instans Anda atau bergabung dengan grup DingTalk Hologres untuk meminta peningkatan. Untuk informasi lebih lanjut, lihat Bagaimana cara mendapatkan dukungan online lebih lanjut?.

Prasyarat

Gunakan lingkungan Spark yang mendukung perintah spark-sql, spark-shell, atau pyspark. Disarankan menggunakan Spark 3.3.0 atau lebih baru untuk menghindari masalah dependensi dan memperoleh akses ke fitur tambahan.
- Anda dapat menggunakan Alibaba Cloud EMR Spark untuk menyiapkan lingkungan Spark secara cepat dan menghubungkannya ke instans Hologres. Untuk informasi lebih lanjut, lihat Fitur EMR Spark.
- Atau, Anda dapat menyiapkan lingkungan Spark independen. Untuk informasi lebih lanjut, lihat Apache Spark.
Untuk membaca dari dan menulis ke Hologres dengan Spark, Anda memerlukan konektor hologres-connector-spark-3.x. Topik ini menggunakan versi 1.5.2 sebagai contoh, yang dapat diunduh dari Maven Central Repository. Konektor ini bersifat open source. Untuk informasi lebih lanjut, lihat Hologres-Connectors.

Untuk mengembangkan dan melakukan debugging lokal pekerjaan Spark dalam Java dengan IDE seperti IntelliJ IDEA, tambahkan dependensi Maven berikut ke file pom.xml Anda.

<dependency>
    <groupId>com.alibaba.hologres</groupId>
    <artifactId>hologres-connector-spark-3.x</artifactId>
    <version>1.5.2</version>
    <classifier>jar-with-dependencies</classifier>
</dependency>

Katalog Hologres

Konektor Hologres versi 1.5.2 dan lebih baru mendukung katalog Hologres, yang memungkinkan penggunaan tabel eksternal untuk membaca dari dan menulis ke Hologres.

Setiap katalog Hologres di Spark dipetakan ke database di Hologres, sedangkan setiap namespace di katalog Hologres dipetakan ke skema di database yang sesuai. Bagian berikut menjelaskan cara menggunakan katalog Hologres di Spark.

Catatan

Katalog Hologres tidak mendukung pembuatan tabel.

Topik ini menggunakan database dan tabel berikut di instans Hologres:

test_db -- Database
  public.test_table1 -- Tabel di skema public
  public.test_table2
  test_schema.test_table3  -- Tabel di skema test_schema

Inisialisasi katalog Hologres

Jalankan spark-sql di kluster Spark, muat konektor Hologres, dan tentukan parameter katalog.

spark-sql --jars hologres-connector-spark-3.x-1.5.2-jar-with-dependencies.jar \
--conf spark.sql.catalog.hologres_external_test_db=com.alibaba.hologres.spark3.HoloTableCatalog \
--conf spark.sql.catalog.hologres_external_test_db.username=*** \
--conf spark.sql.catalog.hologres_external_test_db.password=*** \
--conf spark.sql.catalog.hologres_external_test_db.jdbcurl=jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/test_db

Perintah katalog Hologres

Muat Katalog Hologres

Katalog Hologres di Spark dipetakan ke database Hologres. Pemetaan ini berlaku selama sesi berlangsung.
```
USE hologres_external_test_db;
```
Kueri semua namespace

Namespace di Spark dipetakan ke skema di Hologres. Skema default adalah public. Gunakan perintah USE untuk mengubah skema default.
```
-- Lihat semua namespace di Katalog Hologres, yang sesuai dengan skema di database Hologres.
SHOW NAMESPACES;
```

Kueri tabel dalam namespace

Kueri semua tabel
```
SHOW TABLES;
```

Kueri tabel dalam namespace tertentu

USE test_schema;
SHOW TABLES;

-- Atau, gunakan pernyataan berikut. 
SHOW TABLES IN test_schema;

Baca dari dan tulis ke tabel

Gunakan pernyataan SELECT dan INSERT untuk membaca dari dan menulis ke tabel eksternal.

-- Baca dari tabel.
SELECT * FROM public.test_table1;

-- Tulis ke tabel.
INSERT INTO test_schema.test_table3 SELECT * FROM public.test_table1;

Impor data ke Hologres

Data uji pada bagian ini berasal dari tabel customer dalam dataset TPC-H. Spark dapat membaca data dari file CSV dan menuliskannya ke tabel Hologres. Anda dapat mengunduh data sampel customer. Pernyataan SQL berikut membuat customer_holo_table.

CREATE TABLE customer_holo_table
(
  c_custkey    BIGINT ,
  c_name       TEXT   ,
  c_address    TEXT   ,
  c_nationkey  INT    ,
  c_phone      TEXT   ,
  c_acctbal    DECIMAL(15,2) ,
  c_mktsegment TEXT   ,
  c_comment    TEXT
);

Impor menggunakan Spark-SQL

Di Spark-SQL, penggunaan katalog untuk memuat metadata tabel Hologres lebih mudah digunakan. Anda juga dapat mendeklarasikan tabel Hologres dengan membuat tabel sementara.

Catatan

Versi konektor Spark Hologres sebelum 1.5.2 tidak mendukung katalog. Anda hanya dapat mendeklarasikan tabel Hologres dengan membuat tabel sementara.
Untuk informasi lebih lanjut tentang parameter konektor Spark Hologres, lihat parameter.

Inisialisasi katalog Hologres.

spark-sql --jars hologres-connector-spark-3.x-1.5.2-jar-with-dependencies.jar \
--conf spark.sql.catalog.hologres_external_test_db=com.alibaba.hologres.spark3.HoloTableCatalog \
--conf spark.sql.catalog.hologres_external_test_db.username=*** \
--conf spark.sql.catalog.hologres_external_test_db.password=*** \
--conf spark.sql.catalog.hologres_external_test_db.jdbcurl=jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/test_db

Impor data dari sumber CSV ke tabel Hologres.

Catatan

Sintaksis INSERT INTO di Spark tidak mendukung penggunaan column_list untuk menentukan subset kolom. Misalnya, Anda tidak dapat menggunakan INSERT INTO hologresTable(c_custkey) SELECT c_custkey FROM csvTable untuk menulis data hanya ke bidang c_custkey.

Jika Anda ingin menulis data ke bidang tertentu, gunakan pernyataan CREATE TEMPORARY VIEW untuk mendeklarasikan tampilan sementara Hologres yang hanya berisi bidang yang diperlukan.

Menggunakan katalog

-- Muat katalog Hologres.
USE hologres_external_test_db;

-- Buat sumber data CSV.
CREATE TEMPORARY VIEW csvTable (
    c_custkey BIGINT,
    c_name STRING,
    c_address STRING,
    c_nationkey INT,
    c_phone STRING,
    c_acctbal DECIMAL(15, 2),
    c_mktsegment STRING,
    c_comment STRING)
USING csv OPTIONS (
    path "resources/customer", sep "," -- Untuk pengujian lokal, gunakan jalur mutlak ke file.
);

-- Tulis data dari tabel CSV ke Hologres.
INSERT INTO public.customer_holo_table SELECT * FROM csvTable;

Menggunakan tampilan sementara

-- Buat sumber data CSV.
CREATE TEMPORARY VIEW csvTable (
    c_custkey BIGINT,
    c_name STRING,
    c_address STRING,
    c_nationkey INT,
    c_phone STRING,
    c_acctbal DECIMAL(15, 2),
    c_mktsegment STRING,
    c_comment STRING)
USING csv OPTIONS (
    path "resources/customer", sep ","
);

-- Buat tampilan sementara Hologres.
CREATE TEMPORARY VIEW hologresTable (
    c_custkey BIGINT,
    c_name STRING,
    c_phone STRING)
USING hologres OPTIONS (
    jdbcurl "jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/test_db",
    username "***", 
    password "***", 
    table "customer_holo_table"
);

INSERT INTO hologresTable SELECT c_custkey,c_name,c_phone FROM csvTable;

Impor menggunakan DataFrame

Anda dapat menggunakan tool seperti spark-shell atau pyspark untuk mengembangkan pekerjaan Spark dan memanggil API write guna menulis data. Pekerjaan tersebut membaca data dari file CSV, mengonversinya menjadi DataFrame, lalu menulis DataFrame tersebut ke instans Hologres. Bagian berikut menyediakan contoh kode untuk berbagai bahasa pemrograman. Untuk informasi lebih lanjut tentang parameter konektor Spark Hologres, lihat parameter.

Scala

import org.apache.spark.sql.types._
import org.apache.spark.sql.SaveMode

// Skema sumber CSV.
val schema = StructType(Array(
  StructField("c_custkey", LongType),
  StructField("c_name", StringType),
  StructField("c_address", StringType),
  StructField("c_nationkey", IntegerType),
  StructField("c_phone", StringType),
  StructField("c_acctbal", DecimalType(15, 2)),
  StructField("c_mktsegment", StringType),
  StructField("c_comment", StringType)
))

// Baca data dari file CSV ke dalam DataFrame.
val csvDf = spark.read.format("csv").schema(schema).option("sep", ",").load("resources/customer")

// Tulis DataFrame ke Hologres.
csvDf.write
.format("hologres")
.option("username", "***")
.option("password", "***")
.option("jdbcurl", "jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/test_db")
.option("table", "customer_holo_table")
.mode(SaveMode.Append)
.save()

Java

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.*;
import org.apache.spark.sql.SaveMode;
import java.util.Arrays;
import java.util.List;

public class SparkTest {
    public static void main(String[] args) {
        // Skema sumber CSV.
        List<StructField> asList =
                Arrays.asList(
                        DataTypes.createStructField("c_custkey", DataTypes.LongType, true),
                        DataTypes.createStructField("c_name", DataTypes.StringType, true),
                        DataTypes.createStructField("c_address", DataTypes.StringType, true),
                        DataTypes.createStructField("c_nationkey", DataTypes.IntegerType, true),
                        DataTypes.createStructField("c_phone", DataTypes.StringType, true),
                        DataTypes.createStructField("c_acctbal", new DecimalType(15, 2), true),
                        DataTypes.createStructField("c_mktsegment", DataTypes.StringType, true),
                        DataTypes.createStructField("c_comment", DataTypes.StringType, true));
        StructType schema = DataTypes.createStructType(asList);

        // Jalankan dalam mode lokal.
        SparkSession spark = SparkSession.builder()
                .appName("Spark CSV Example")
                .master("local[*]") 
                .getOrCreate();

        // Baca data dari file CSV ke dalam DataFrame.
        // Untuk pengujian lokal, gunakan jalur mutlak ke data customer.
        Dataset<Row> csvDf = spark.read().format("csv").schema(schema).option("sep", ",").load("resources/customer");

        // Tulis DataFrame ke Hologres.
        csvDf.write.format("hologres").option(
           "username", "***").option(
           "password", "***").option(
           "jdbcurl", "jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/test_db").option(
           "table", "customer_holo_table").mode(
           "append").save();
    }
}

Tambahkan dependensi berikut ke file pom.xml Anda.

<dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-sql_2.13</artifactId>
      <version>3.5.4</version>
      <scope>provided</scope>
</dependency>

Python

from pyspark.sql.types import *

# Skema sumber CSV.
schema = StructType([
    StructField("c_custkey", LongType()),
    StructField("c_name", StringType()),
    StructField("c_address", StringType()),
    StructField("c_nationkey", IntegerType()),
    StructField("c_phone", StringType()),
    StructField("c_acctbal", DecimalType(15, 2)),
    StructField("c_mktsegment", StringType()),
    StructField("c_comment", StringType())
])

# Baca data dari file CSV ke dalam DataFrame.
csvDf = spark.read.csv("resources/customer", header=False, schema=schema, sep=',')

# Tulis DataFrame ke Hologres.
csvDf.write.format("hologres").option(
    "username", "***").option(
    "password", "***").option(
    "jdbcurl", "jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/test_db").option(
    "table", "customer_holo_table").mode(
    "append").save()

Untuk menjalankan pekerjaan Spark dalam berbagai bahasa, ikuti langkah-langkah berikut:

Scala
- Gunakan kode contoh untuk membuat file sparktest.scala dan jalankan perintah berikut untuk mengeksekusi pekerjaan.
```
-- Muat dependensi. 
spark-shell --jars hologres-connector-spark-3.x-1.5.2-jar-with-dependencies.jar

-- Untuk pengujian lokal, gunakan jalur mutlak untuk memuat file.
scala> :load D:/sparktest.scala
```
- Atau, Anda dapat menempelkan kode contoh langsung ke shell untuk mengeksekusinya setelah memuat dependensi.
Java

Gunakan tool pengembangan untuk mengimpor kode contoh dan mengemasnya dengan Maven. Misalnya, jika file JAR keluaran adalah spark_test.jar, jalankan perintah berikut untuk mengeksekusi pekerjaan.
```
-- Gunakan jalur mutlak ke paket JAR pekerjaan.
spark-submit --class SparkTest  --jars hologres-connector-spark-3.x-1.5.2-jar-with-dependencies.jar  D:\spark_test.jar
```
Python

Setelah menjalankan perintah berikut, Anda dapat menempelkan kode contoh langsung ke shell untuk mengeksekusinya.
```
pyspark --jars hologres-connector-spark-3.x-1.5.2-jar-with-dependencies.jar
```

Baca data dari Hologres

Mulai dari versi 1.3.2, konektor Spark mendukung pembacaan data dari Hologres. Dibandingkan dengan jdbc-connector Spark default, spark-connector memberikan kinerja lebih baik dengan membaca data secara paralel berdasarkan shard tabel Hologres. Paralelisme baca terkait dengan jumlah shard di tabel. spark-connector dapat membatasi paralelisme menggunakan parameter read.max_task_count. Pekerjaan akhirnya menghasilkan Min(shardCount, max_task_count) tugas baca. Konektor ini juga mendukung inferensi skema. Jika Anda tidak menyediakan skema, konektor akan menginferensi skema Spark dari skema tabel Hologres.
Mulai dari versi konektor Spark 1.5.0, pembacaan data dari tabel Hologres mendukung predicate pushdown, LIMIT pushdown, dan column pruning. Anda juga dapat menggunakan SELECT QUERY Hologres untuk membaca data. Versi ini memperkenalkan mode baca batch, yang meningkatkan kinerja baca hingga 3–4 kali lipat dibandingkan versi sebelumnya.

Baca data menggunakan Spark SQL

Saat menggunakan Spark SQL, Anda dapat memuat metadata tabel Hologres menggunakan katalog. Atau, Anda dapat mendeklarasikan tabel Hologres dengan membuat tabel sementara.

Catatan

Versi konektor Spark Hologres sebelum 1.5.2 tidak mendukung katalog. Anda hanya dapat mendeklarasikan tabel Hologres dengan membuat tabel sementara.
Untuk informasi lebih lanjut tentang parameter konektor Spark Hologres, lihat Parameter.

Inisialisasi katalog Hologres.

spark-sql --jars hologres-connector-spark-3.x-1.5.2-jar-with-dependencies.jar \
--conf spark.sql.catalog.hologres_external_test_db=com.alibaba.hologres.spark3.HoloTableCatalog \
--conf spark.sql.catalog.hologres_external_test_db.username=*** \
--conf spark.sql.catalog.hologres_external_test_db.password=*** \
--conf spark.sql.catalog.hologres_external_test_db.jdbcurl=jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/test_db

Baca data dari Hologres.

Baca data menggunakan katalog.

-- Muat katalog Hologres.
USE hologres_external_test_db;

-- Baca data dari tabel Hologres. Field pruning dan predicate pushdown didukung.
SELECT c_custkey,c_name,c_phone FROM public.customer_holo_table WHERE c_custkey < 500 LIMIT 10;

Baca data dengan membuat tabel sementara.

Tabel

CREATE TEMPORARY VIEW hologresTable
USING hologres OPTIONS (
  jdbcurl "jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/test_db",
  username "***", 
  password "***", 
  read.max_task_count "80", -- Jumlah maksimum tugas untuk membaca dari tabel Hologres.
  table "customer_holo_table"
);

-- Field pruning dan predicate pushdown didukung.
SELECT c_custkey,c_name,c_phone FROM hologresTable WHERE c_custkey < 500 LIMIT 10;

Kueri

CREATE TEMPORARY VIEW hologresTable
USING hologres OPTIONS (
  jdbcurl "jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/test_db",
  username "***", 
  password "***", 
  read.query "SELECT c_custkey,c_name,c_phone FROM customer_holo_table WHERE c_custkey < 500 LIMIT 10"
);

SELECT * FROM hologresTable LIMIT 5;

Baca data Hologres ke dalam DataFrame

Saat mengembangkan pekerjaan Spark menggunakan tool seperti spark-shell atau pyspark, Anda dapat memanggil API read Spark untuk memuat data ke dalam DataFrame. Contoh berikut menunjukkan cara membaca data dari tabel Hologres ke dalam DataFrame dalam berbagai bahasa pemrograman. Untuk informasi lebih lanjut tentang parameter konektor Spark Hologres, lihat Parameter.

Scala

val readDf = (
  spark.read
    .format("hologres")
    .option("username", "***")
    .option("password", "***")
    .option("jdbcurl", "jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/test_db")
    .option("table", "customer_holo_table")
    .option("read.max_task_count", "80") // Jumlah maksimum tugas untuk membaca dari tabel Hologres.
    .load()
    .filter("c_custkey < 500")
)

readDf.select("c_custkey", "c_name", "c_phone").show(10)

Java

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

public class SparkSelect {
    public static void main(String[] args) {
        
        // Jalankan dalam mode lokal.
        SparkSession spark = SparkSession.builder()
                .appName("Spark CSV Example")
                .master("local[*]") 
                .getOrCreate();
                
        Dataset<Row> readDf = (
           spark.read
                .format("hologres")
                .option("username", "***")
                .option("password", "***")
                .option("jdbcurl", "jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/test_db")
                .option("table", "customer_holo_table")
                .option("read.max_task_count", "80") // Jumlah maksimum tugas untuk membaca dari tabel Hologres.
                .load()
                .filter("c_custkey < 500")
        );
        readDf.select("c_custkey", "c_name", "c_phone").show(10);
    }
}

Tambahkan dependensi berikut ke file Maven pom.xml Anda.

<dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-sql_2.13</artifactId>
      <version>3.5.4</version>
      <scope>provided</scope>
</dependency>

Python

readDf = spark.read.format("hologres").option(
"username", "***").option(
"password", "***").option(
"jdbcurl", "jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/test_db").option(
"table", "customer_holo_table").option(
"read.max_task_count", "80").load()

readDf.select("c_custkey", "c_name", "c_phone").show(10)

Untuk menjalankan pekerjaan Spark dalam berbagai bahasa pemrograman:

Scala
- Anda dapat menggunakan kode contoh untuk membuat file sparkselect.scala dan menjalankan pekerjaan dengan perintah berikut.
```
-- Muat dependensi.
spark-shell --jars hologres-connector-spark-3.x-1.5.2-jar-with-dependencies.jar

-- Untuk pengujian lokal, gunakan jalur mutlak untuk memuat file.
scala> :load D:/sparkselect.scala
```
- Atau, setelah dependensi dimuat, Anda dapat menempelkan kode contoh langsung ke shell untuk menjalankannya.
Java

Anda dapat menggunakan tool pengembangan untuk mengimpor kode contoh dan mengemasnya menggunakan tool Maven. Misalnya, jika paket JAR bernama spark_select.jar, jalankan pekerjaan dengan perintah berikut.
```
-- Gunakan jalur mutlak untuk JAR pekerjaan.
spark-submit --class SparkSelect --jars hologres-connector-spark-3.x-1.5.2-jar-with-dependencies.jar  D:\spark_select.jar
```
Python

Setelah menjalankan perintah berikut, Anda dapat menempelkan kode contoh langsung ke shell untuk menjalankannya.
```
pyspark --jars hologres-connector-spark-3.x-1.5.2-jar-with-dependencies.jar
```

Parameter

Parameter umum

Parameter	Default	Wajib	Deskripsi
username	None	Ya	ID AccessKey Akun Alibaba Cloud Anda. Anda dapat memperolehnya di halaman AccessKey Management. Atau, nama akun pengguna. Setelah membuat akun, berikan izin database yang diperlukan. Untuk informasi lebih lanjut, lihat model izin Hologres. Kemudian, hubungkan ke HoloWeb dan jalankan kueri untuk memverifikasi izin.
password	None	Ya	Secret AccessKey yang sesuai dengan ID AccessKey Anda. Untuk informasi lebih lanjut, lihat Buat AccessKey. Atau, kata sandi untuk akun tersebut.
table	None	Ya	Nama tabel Hologres yang akan dibaca atau ditulis. Catatan Saat membaca data, Anda dapat menggunakan parameter `read.query` sebagai gantinya.
jdbcurl	None	Ya	URL JDBC API data real-time Hologres, dalam format `jdbc:postgresql://<host>:<port>/<db_name>`. Untuk menemukan detail ini, buka Hologres console, klik Instances di panel navigasi kiri, pilih instans target Anda, lalu temukan host dan nomor port di bagian Network Information pada halaman Instance Details.
enable_serverless_computing	false	Tidak	Menentukan apakah akan menggunakan resource komputasi tanpa server. Parameter ini hanya berlaku untuk operasi baca dan operasi tulis dalam mode `bulk_load`. Untuk informasi lebih lanjut, lihat Panduan pengguna Komputasi Tanpa Server.
serverless_computing_query_priority	3	Tidak	Prioritas eksekusi untuk Komputasi Tanpa Server.
statement_timeout_seconds	28800 (8 jam)	Tidak	Durasi timeout eksekusi kueri, dalam detik.
retry_count	3	Tidak	Jumlah percobaan ulang jika koneksi gagal.
direct_connect	Koneksi langsung digunakan secara default jika didukung.	Tidak	Menentukan apakah akan terhubung langsung ke FrontEnd Hologres (node akses). Throughput jaringan titik akhir sering menjadi bottleneck untuk operasi data batch. Secara default, konektor menggunakan koneksi langsung jika tersedia untuk meningkatkan throughput. Tetapkan nilai ini ke `false` untuk menonaktifkan perilaku ini.

Parameter tulis

Konektor Hologres mendukung parameter Spark SaveMode. Untuk SQL, ini sesuai dengan INSERT INTO atau INSERT OVERWRITE. Untuk DataFrame, Anda dapat mengatur SaveMode ke Append atau Overwrite saat menulis data. Mode Overwrite membuat tabel sementara untuk operasi tulis dan menggantikan tabel asli setelah berhasil. Gunakan mode ini hanya jika diperlukan.

Parameter	Nama sebelumnya	Default	Wajib	Deskripsi
write.mode	copy_write_mode	auto	Tidak	Mode tulis. Untuk perbandingan mode tulis, lihat Mode tulis batch. Nilai yang valid: `auto` (Default): Konektor secara otomatis memilih mode optimal berdasarkan versi Hologres dan metadata tabel tujuan. Logika pemilihan adalah sebagai berikut: Jika instans Hologres v2.2.25 atau lebih baru dan tabel memiliki primary key, mode `bulk_load_on_conflict` digunakan. Jika instans Hologres v2.1.0 atau lebih baru dan tabel tidak memiliki primary key, mode `bulk_load` digunakan. Jika instans Hologres v1.3 atau lebih baru, mode `stream` digunakan. Dalam kasus lain, mode `insert` digunakan. `stream`: Menggunakan Fixed Plan untuk mempercepat eksekusi SQL. Dalam Fixed Plan, `COPY` adalah fitur yang diperkenalkan di Hologres v1.3. Dibandingkan metode `INSERT`, `COPY` memberikan throughput lebih tinggi (karena sifat streaming-nya), latensi data lebih rendah, dan konsumsi memori klien berkurang (karena tidak membatch data). Catatan Memerlukan konektor Hologres v1.3.0 atau lebih baru dan Hologres v1.3.34 atau lebih baru. `bulk_load`: Menggunakan `COPY` batch. Dibandingkan `COPY` streaming dalam Fixed Plan, `COPY` batch memberikan beban lebih rendah pada instans Hologres dalam kondisi RPS tinggi tetapi hanya mendukung penulisan ke tabel tanpa primary key. Catatan Memerlukan konektor Hologres v1.4.2 atau lebih baru dan Hologres v2.1.0 atau lebih baru. `bulk_load_on_conflict`: Menggunakan `COPY` batch untuk menulis ke tabel dengan primary key dan dapat menangani primary key duplikat. Secara default, impor data batch ke tabel Hologres dengan primary key memicu penguncian tabel, yang membatasi operasi tulis konkuren dari beberapa koneksi. Konektor dapat mendistribusikan ulang data berdasarkan kunci distribusi tabel tujuan, sehingga setiap tugas Spark menulis ke satu shard saja. Hal ini mengurangi penguncian tabel menjadi penguncian tingkat shard, memungkinkan penulisan konkuren dan meningkatkan kinerja tulis. Karena setiap koneksi hanya menyimpan data untuk beberapa shard, optimasi ini juga secara signifikan mengurangi jumlah file kecil dan menurunkan penggunaan memori Hologres. Pengujian menunjukkan bahwa repartisi data sebelum penulisan konkuren dapat mengurangi beban sistem sekitar 67% dibandingkan mode `stream`. Catatan Memerlukan konektor Hologres v1.4.2 atau lebih baru dan Hologres v2.2.25 atau lebih baru. `insert`: Menulis data menggunakan metode `INSERT`. `stage`: Menggunakan tabel Stage untuk penulisan batch. Mode ini mengimpor data secara near-real time melalui stage penyimpanan sementara lalu memuatnya ke tabel tujuan menggunakan pernyataan `INSERT FROM SELECT`. Dalam skenario batch, mode ini menawarkan kinerja lebih baik dan beban sistem lebih rendah dibandingkan mode lain. Mode ini juga mendukung penggunaan resource komputasi tanpa server. Operasi `INSERT INTO` dan `INSERT OVERWRITE` didukung. Memerlukan Hologres v4.1.0 atau lebih baru dan konektor v1.6.1 atau lebih baru.
write.copy.max_buffer_size	max_cell_buffer_size	52428800 (50 MB)	Tidak	Ukuran maksimum buffer lokal saat menulis dalam mode `COPY`. Biasanya Anda tidak perlu menyesuaikan nilai ini. Namun, Anda dapat menaikkan nilai ini jika terjadi luapan buffer saat menulis field besar, seperti string yang sangat panjang.
write.copy.dirty_data_check	copy_write_dirty_data_check	false	Tidak	Menentukan apakah akan memeriksa data kotor. Jika diaktifkan, fitur ini dapat mengidentifikasi baris yang gagal ditulis secara tepat. Namun, hal ini memengaruhi kinerja tulis. Nonaktifkan fitur ini kecuali untuk troubleshooting.
write.on_conflict_action	INSERT_OR_REPLACE	INSERT_OR_REPLACE	Tidak	Aksi yang diambil ketika operasi tulis mengalami konflik primary key di tabel tujuan. INSERT_OR_IGNORE: Jika terjadi konflik primary key, data tidak ditulis. INSERT_OR_UPDATE: Jika terjadi konflik primary key, kolom yang ditentukan diperbarui. INSERT_OR_REPLACE: Jika terjadi konflik primary key, semua kolom diperbarui.
write.stage.compression	None	false	Tidak	Berlaku saat `write.mode` diatur ke `stage`. Mengatur nilai ini ke `true` mengaktifkan kompresi selama penulisan stage. Memerlukan Hologres v4.2.8 atau lebih baru dan konektor v1.6.2 atau lebih baru.

Parameter berikut hanya berlaku saat write.mode diatur ke insert.

Parameter	Nama sebelumnya	Default	Wajib	Deskripsi
write.insert.dynamic_partition	dynamic_partition	false	Tidak	Saat `write.mode` adalah `insert`, mengatur nilai ini ke `true` secara otomatis membuat partisi yang belum ada saat menulis ke tabel partisi induk.
write.insert.batch_size	write_batch_size	512	Tidak	Ukuran batch maksimum untuk setiap thread tulis. Commit batch dipicu ketika jumlah operasi `Put` mencapai nilai ini.
write.insert.batch_byte_size	write_batch_byte_size	2097152 (2 MB)	Tidak	Ukuran batch maksimum dalam byte untuk setiap thread tulis. Nilai default adalah 2 MB. Commit batch dipicu ketika ukuran byte data `Put` mencapai nilai ini.
write.insert.max_interval_ms	write_max_interval_ms	10000	Tidak	Commit batch dipicu jika waktu yang berlalu sejak commit terakhir melebihi nilai ini.
write.insert.thread_size	write_thread_size	1	Tidak	Jumlah thread tulis konkuren. Setiap thread menggunakan satu koneksi database.
write.rps_limit	None	-1	Tidak	Batas laju tulis per tugas, dalam baris per detik (RPS). Nilai default -1 menunjukkan tidak ada batas.

Parameter baca

Parameter	Nama sebelumnya (v1.5.0 dan sebelumnya)	Default	Wajib	Deskripsi
read.mode	bulk_read	auto	Tidak	Mode baca. Nilai yang valid: `auto` (Default): Konektor secara otomatis memilih mode optimal berdasarkan versi Hologres dan metadata tabel. Logika pemilihan adalah sebagai berikut: Jika bidang yang akan dibaca mencakup tipe data `JSONB`, mode `select` digunakan. Jika instans v3.0.24 atau lebih baru, mode bulk_read_compressed digunakan. Dalam kasus lain, mode bulk_read digunakan. `bulk_read`: Menggunakan metode `COPY OUT` untuk membaca data dalam format `arrow`, yang beberapa kali lebih cepat daripada mode `select`. Membaca tipe `JSONB` dari Hologres saat ini tidak didukung. `bulk_read_compressed`: Membaca data terkompresi dalam format `Arrow` menggunakan metode COPY OUT, yang menghemat sekitar 45% bandwidth dibandingkan membaca data tidak terkompresi. `select`: Menggunakan pernyataan `SELECT` standar untuk membaca data.
read.max_task_count	max_partition_count	80	Tidak	Menentukan jumlah maksimum tugas konkuren untuk membaca data. Konektor membagi tabel menjadi beberapa partisi, dan setiap partisi diproses oleh satu tugas Spark. Jika jumlah shard tabel kurang dari nilai ini, jumlah partisi dibatasi pada jumlah shard.
read.copy.max_buffer_size	/	52428800 (50 MB)	Tidak	Ukuran maksimum buffer lokal saat membaca dalam mode `COPY`. Jika terjadi exception karena ukuran field besar, naikkan nilai ini.
read.push_down_predicate	push_down_predicate	true	Tidak	Menentukan apakah akan mengaktifkan predicate pushdown. Saat diaktifkan, operasi seperti kondisi filter dan column pruning didorong ke sumber data.
read.push_down_limit	push_down_limit	true	Tidak	Menentukan apakah akan mengaktifkan pushdown `LIMIT`.
read.select.batch_size	scan_batch_size	256	Tidak	Berlaku saat `read.mode` diatur ke `select`. Menentukan jumlah baris yang diambil dalam satu operasi pemindaian saat membaca dari Hologres.
read.select.timeout_seconds	scan_timeout_seconds	60	Tidak	Berlaku saat `read.mode` diatur ke `select`. Menentukan durasi timeout untuk operasi pemindaian saat membaca dari Hologres.
read.query	query	None	Tidak	Gunakan `query` yang disediakan untuk membaca dari Hologres. Parameter ini tidak dapat digunakan bersama parameter `table`. Catatan Saat membaca data menggunakan metode `query`, Anda hanya dapat menggunakan satu tugas. Predicate pushdown tidak didukung. Saat membaca data menggunakan metode `table`, operasi baca dibagi menjadi beberapa Tugas yang berjalan secara paralel berdasarkan ShardCount tabel Hologres.
read.split.strategy	None	auto	Tidak	Menentukan strategi untuk membagi data tabel menjadi beberapa tugas untuk pembacaan paralel. Nilai yang valid: `auto` (default): Konektor secara otomatis memilih strategi optimal. `shard`: Membagi data berdasarkan shard di Hologres, dan setiap tugas membaca sebagian shard. Cocok untuk tabel dengan distribusi shard yang jelas. `range`: Membagi data berdasarkan rentang nilai kolom tertentu. Pengaturan ini memerlukan parameter `read.split.column`, `read.split.lower_bound`, `read.split.upper_bound`, dan `read.split.num`. `partition` membagi data berdasarkan nilai kolom partisi. Opsi ini harus digunakan dengan parameter `read.split.column` dan `read.split.num`. Berlaku untuk tabel partisi atau tampilan partisi. Catatan Didukung di konektor v1.6.1 dan lebih baru.
read.split.column	None	None	Tidak	Nama kolom pemisah. Parameter ini wajib saat `read.split.strategy` adalah `range` atau `partition`.
read.split.lower_bound	None	None	Tidak	Batas bawah pemisah wajib saat `read.split.strategy` diatur ke `range`. Menentukan nilai batas bawah untuk pemisahan rentang.
read.split.upper_bound	None	None	Tidak	Batas atas pemisah menentukan batas atas untuk shard berbasis rentang. Parameter ini wajib saat `read.split.strategy` diatur ke `range`.
read.split.num	None	None	Tidak	Jumlah pemisahan. Parameter ini wajib saat `read.split.strategy` adalah `range` atau `partition`. Menentukan jumlah tugas yang membagi data untuk dibaca.

Pemetaan tipe data

Tipe Spark	Tipe Hologres
ShortType	SMALLINT
IntegerType	INT
LongType	BIGINT
StringType	TEXT
StringType	JSON
StringType	JSONB
DecimalType	NUMERIC(38, 18)
BooleanType	BOOL
DoubleType	DOUBLE PRECISION
FloatType	FLOAT
TimestampType	TIMESTAMPTZ
DateType	DATE
BinaryType	BYTEA
BinaryType	ROARINGBITMAP
ArrayType(IntegerType)	INT4[]
ArrayType(LongType)	INT8[]
ArrayType(FloatType)	FLOAT4[]
ArrayType(DoubleType)	FLOAT8[]
ArrayType(BooleanType)	BOOLEAN[]
ArrayType(StringType)	TEXT[]

Perhitungan jumlah koneksi

Hologres-Connector-Spark menggunakan sejumlah koneksi JDBC tertentu untuk operasi baca dan tulis. Jumlah koneksi tergantung pada faktor-faktor berikut:

Paralelisme Spark, yaitu jumlah tugas konkuren yang berjalan selama eksekusi pekerjaan, yang terlihat di Spark UI.
Jumlah koneksi yang digunakan per tugas:
- Saat menulis dalam mode COPY, setiap tugas menggunakan satu koneksi JDBC.
- Saat menulis dalam mode INSERT, setiap tugas menggunakan write_thread_size koneksi JDBC.
- Saat membaca data, setiap tugas menggunakan satu koneksi JDBC.
Operasi lain: Pekerjaan mungkin secara singkat menggunakan satu koneksi untuk tugas seperti pengambilan skema saat dimulai.

Anda dapat menghitung total jumlah koneksi untuk pekerjaan dengan rumus berikut:

Item	Koneksi
Mengkueri metadata dari katalog	1
Membaca data	parallelism * 1 + 1
Menulis dalam mode COPY	parallelism * 1 + 1
Menulis dalam mode INSERT	parallelism * write_thread_size + 1

Perhitungan ini mengasumsikan bahwa kapasitas Spark untuk tugas konkuren melebihi jumlah tugas yang dihasilkan pekerjaan.

Kapasitas Spark untuk tugas konkuren tergantung pada parameter yang dikonfigurasi pengguna, seperti spark.executor.instances, dan kebijakan pemisahan blok file Hadoop. Untuk informasi lebih lanjut, lihat Apache Hadoop.

Catatan rilis Spark Connector

Versi	Tanggal rilis	Fitur baru	Perbaikan bug
1.6.1	2026-02	Menambahkan mode tulis stage, yang menggunakan tabel stage sementara untuk meningkatkan kinerja tulis. Fitur ini memerlukan Hologres v4.1.0 atau lebih baru. Menambahkan dukungan untuk operasi `INSERT INTO` dan `INSERT OVERWRITE` dalam mode stage. Mengoptimalkan logika pembersihan tabel sementara untuk operasi `overwrite`. Menambahkan parameter `write.rps_limit` untuk mengontrol laju tulis per tugas.	Memperbaiki masalah replay sinkron paksa yang terjadi saat mengeksekusi DDL selama operasi `overwrite`.
1.6.0	2025-12	Menambahkan parameter `read.split.strategy` untuk mendukung tiga strategi pemisahan baca: `shard`, `range`, dan `partition`. Menambahkan dukungan untuk tipe data `JSONB` dalam pembacaan berbasis Arrow. Menambahkan parameter `write.insert.ignore_null_when_update` untuk mengabaikan nilai NULL selama operasi `UPDATE`. Menambahkan dukungan untuk pelaporan cakupan pengujian JaCoCo.	Memperbaiki error saat menulis nilai `date` sebelum tahun 1970.
1.5.6	2025-11	Menambahkan dukungan untuk metode autentikasi AKV4. Menambahkan parameter `write.copy.disable_right_join` untuk mengoptimalkan kinerja tulis `COPY`.	Mengoptimalkan logika pemeriksaan tipe kolom untuk penulisan katalog.
1.5.5	2025-10	Menambahkan dukungan untuk menulis data presisi rendah ke tipe data presisi tinggi. Menambahkan awalan seperti `appname` dan `taskid` ke log untuk menyederhanakan troubleshooting. Mengaktifkan kompresi Arrow LZ4 untuk operasi `COPY OUT`.	Pemformatan dan validasi parameter telah dioptimalkan.
1.5.4	2025-09	Menambahkan parameter `remove_u0000` untuk secara otomatis menghapus karakter `U+0000` dari kolom teks. Merombak Hologres Catalog untuk memetakan namespace ke skema. Menambahkan kelas utilitas `RepartitionUtil` untuk sharding data.	Merombak dokumentasi dan memperbarui tautan.
1.5.2	2025-08	Memperkenalkan Hologres Catalog, yang memungkinkan pembacaan dari dan penulisan ke Hologres menggunakan tabel eksternal. Menambahkan dukungan untuk predicate pushdown dan limit pushdown. Mengaktifkan mode batch `bulk_read` untuk operasi baca, meningkatkan kinerja hingga satu orde besaran.	Mengoptimalkan kinerja baca.
1.5.0	2025-06	Menambahkan dukungan untuk predicate pushdown, limit pushdown, dan column pruning. Mengaktifkan pembacaan data menggunakan kueri `SELECT`. Mengaktifkan mode batch untuk operasi baca.	Memperbaiki masalah terkait pembacaan.
1.4.2	2025-04	Memperkenalkan mode tulis `bulk_load` untuk penulisan batch ke tabel tanpa primary key. Menambahkan mode `bulk_load_on_conflict` untuk menangani konflik primary key di tabel dengan primary key.	Performa tulis yang telah dioptimalkan.
1.3.2	2025-02	Menambahkan dukungan untuk membaca data dari Hologres. Mengaktifkan pembacaan paralel berdasarkan shard. Menambahkan dukungan untuk inferensi skema otomatis.	Menstabilkan fungsionalitas baca awal.
1.3.0	2025-01	Memperkenalkan mode tulis `stream` (fixed `COPY`). Menambahkan dukungan untuk fitur fixed `COPY` di Hologres v1.3.	Mengoptimalkan throughput dan latensi tulis.