Read and write Hologres with Spark - Hologres - Alibaba Cloud Documentation Center

Spark is a unified analytics engine for large-scale data processing. Hologres integrates efficiently with community Spark and EMR Serverless Spark to quickly build a data warehouse. Hologres's Spark connector supports creating a Hologres catalog in a Spark cluster. This enables high-performance batch reads and imports using external tables, outperforming native JDBC.

Limitations

The Spark connector requires Hologres version 1.3 or later. You can check your instance version on the Instance Details page in the Hologres console. If your instance is older than version 1.3, upgrade your instance or join the Hologres DingTalk group (ID: 32314975) to request an upgrade.

Prerequisites

Use a Spark environment that can run spark-sql, spark-shell, or pyspark commands. Use Spark 3.3.0 or later to avoid dependency issues and access more features.
- You can use Alibaba Cloud EMR Spark to quickly set up a Spark environment and connect to a Hologres instance. For more information, see EMR Spark features.
- Alternatively, you can set up an independent Spark environment. For more information, see Apache Spark.
To read from and write to Hologres with Spark, you need the hologres-connector-spark-3.x connector. This topic uses version 1.5.2 as an example, which you can download from the Maven Central Repository. The connector is open source. For more information, see Hologres-Connectors.

To develop and locally debug Spark jobs in Java with an IDE like IntelliJ IDEA, add the following Maven dependency to your pom.xml file.

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

Hologres catalog

Hologres connector 1.5.2 and later supports Hologres catalogs, allowing you to use external tables to read from and write to Hologres.

Each Hologres catalog in Spark maps to a database in Hologres. Each namespace in a Hologres catalog maps to a schema in the corresponding database. The following sections explain how to use Hologres catalogs in Spark.

Note

Hologres catalogs do not support table creation.

This topic uses the following database and tables in a Hologres instance:

test_db -- Database
  public.test_table1 -- Table in the public schema
  public.test_table2
  test_schema.test_table3  -- Table in the test_schema schema

Initialize a Hologres catalog

Start spark-sql in the Spark cluster, load the Hologres connector, and specify the catalog parameters.

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

Hologres catalog commands

Load a Hologres Catalog
A Hologres catalog in Spark maps to a Hologres database. This mapping is fixed for the duration of the session.
```
USE hologres_external_test_db;
```
Query all namespaces
A namespace in Spark maps to a schema in Hologres. The default schema is public. Use the USE command to change the default schema.
```
-- View all namespaces in the Hologres Catalog, which correspond to the schemas in the Hologres database.
SHOW NAMESPACES;
```

Query tables in a namespace

Query all tables
```
SHOW TABLES;
```

Query tables in a specific namespace

USE test_schema;
SHOW TABLES;

-- Or, use the following statement. 
SHOW TABLES IN test_schema;

Read from and write to a table

Use SELECT and INSERT statements to read from and write to external tables.

-- Read from the table.
SELECT * FROM public.test_table1;

-- Write to the table.
INSERT INTO test_schema.test_table3 SELECT * FROM public.test_table1;

Import data to Hologres

The test data in this section is from the customer table in a TPC-H dataset. Spark can read data from a CSV file and write it to a Hologres table. You can download the sample customer data. The following SQL statement creates the 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
);

Import using Spark-SQL

In Spark-SQL, using a catalog to load Hologres table metadata is more convenient. You can also declare a Hologres table by creating a temporary table.

Note

Hologres Spark connector versions earlier than 1.5.2 do not support catalogs. You can only declare a Hologres table by creating a temporary table.
For more information about the parameters for the Hologres Spark connector, see parameters.

Initialize a Hologres catalog.

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

Import data from a CSV source to a Hologres table.

Note

The INSERT INTO syntax in Spark does not support using column_list to specify a subset of columns. For example, you cannot use INSERT INTO hologresTable(c_custkey) SELECT c_custkey FROM csvTable to write data only to the c_custkey field.

If you want to write data to specific fields, use the CREATE TEMPORARY VIEW statement to declare a Hologres temporary view that contains only the required fields.

Using a catalog

-- Load a Hologres catalog.
USE hologres_external_test_db;

-- Create a CSV data source.
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 "," -- For local testing, use the absolute path to the file.
);

-- Write data from the CSV table to Hologres.
INSERT INTO public.customer_holo_table SELECT * FROM csvTable;

Using a temporary view

-- Create a CSV data source.
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 ","
);

-- Create a Hologres temporary view.
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;

Import using a DataFrame

You can use tools such as spark-shell or pyspark to develop Spark jobs and call the write API to write data. The job reads data from a CSV file, converts it to a DataFrame, and then writes the DataFrame to a Hologres instance. The following sections provide sample code for different programming languages. For more information about the parameters for the Hologres Spark connector, see parameters.

Scala

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

// The schema of the CSV source.
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)
))

// Read data from a CSV file into a DataFrame.
val csvDf = spark.read.format("csv").schema(schema).option("sep", ",").load("resources/customer")

// Write the DataFrame to 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) {
        // The schema of the CSV source.
        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);

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

        // Read data from a CSV file into a DataFrame.
        // For local testing, use the absolute path to the customer data.
        Dataset<Row> csvDf = spark.read().format("csv").schema(schema).option("sep", ",").load("resources/customer");

        // Write the DataFrame to 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();
    }
}

Add the following dependency to your pom.xml file.

<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 *

# The schema of the CSV source.
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())
])

# Read data from a CSV file into a DataFrame.
csvDf = spark.read.csv("resources/customer", header=False, schema=schema, sep=',')

# Write the DataFrame to 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()

To run the Spark jobs in different languages, follow these steps:

Scala
- Use the sample code to create a sparktest.scala file and run the following commands to execute the job.
```
-- Load dependencies. 
spark-shell --jars hologres-connector-spark-3.x-1.5.2-jar-with-dependencies.jar

-- For local testing, use the absolute path to load the file.
scala> :load D:/sparktest.scala
```
- Alternatively, you can paste the sample code directly into the shell to execute it after loading the dependencies.
Java
Use a development tool to import the sample code and package it with Maven. For example, if the output JAR file is spark_test.jar, run the following command to execute the job.
```
-- Use the absolute path to the job JAR package.
spark-submit --class SparkTest  --jars hologres-connector-spark-3.x-1.5.2-jar-with-dependencies.jar  D:\spark_test.jar
```
Python
After running the following command, you can paste the sample code into the shell to execute it.
```
pyspark --jars hologres-connector-spark-3.x-1.5.2-jar-with-dependencies.jar
```

Read data from Hologres

Starting from version 1.3.2, the Spark connector supports reading data from Hologres. Compared with the default Spark jdbc-connector, the spark-connector provides better performance by reading data in parallel based on the shards of a Hologres table. The read parallelism is related to the number of shards in the table. The spark-connector can limit the parallelism by using the read.max_task_count parameter. The job ultimately generates Min(shardCount, max_task_count) read tasks. It also supports schema inference. If you do not provide a schema, it infers the Spark schema from the Hologres table schema.
Starting from Spark connector version 1.5.0, reading data from Hologres tables supports predicate pushdown, LIMIT pushdown, and column pruning. You can also use a Hologres SELECT QUERY to read data. This version introduces batch read mode, which improves read performance by 3 to 4 times compared to previous versions.

Read data using Spark SQL

When using Spark SQL, you can load the metadata of a Hologres table using a catalog. Alternatively, you can declare a Hologres table by creating a temporary table.

Note

Hologres Spark connector versions earlier than 1.5.2 do not support catalogs. You can only declare a Hologres table by creating a temporary table.
For more information about the Hologres Spark connector parameters, see Parameters.

Initialize a Hologres catalog.

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

Read data from Hologres.

Read data using a catalog.

-- Load the Hologres catalog.
USE hologres_external_test_db;

-- Read data from the Hologres table. Field pruning and predicate pushdown are supported.
SELECT c_custkey,c_name,c_phone FROM public.customer_holo_table WHERE c_custkey < 500 LIMIT 10;

Read data by creating a temporary table.

Table

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", -- The maximum number of tasks to use for reading from the Hologres table.
  table "customer_holo_table"
);

-- Field pruning and predicate pushdown are supported.
SELECT c_custkey,c_name,c_phone FROM hologresTable WHERE c_custkey < 500 LIMIT 10;

Query

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;

Read Hologres data into a DataFrame

When you develop Spark jobs using tools such as spark-shell or pyspark, you can call Spark's read API to load data into a DataFrame. The following examples show how to read data from a Hologres table into a DataFrame in different programming languages. For more information about the Hologres Spark connector parameters, see Parameters.

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") // The maximum number of tasks to use for reading from the Hologres table.
    .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) {
        
        // Run in local mode.
        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") // The maximum number of tasks to use for reading from the Hologres table.
                .load()
                .filter("c_custkey < 500")
        );
        readDf.select("c_custkey", "c_name", "c_phone").show(10);
    }
}

Add the following dependency to your Maven pom.xml file.

<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)

To run Spark jobs in different programming languages:

Scala
- You can use the sample code to create a sparkselect.scala file and run the job with the following command.
```
-- Load dependencies.
spark-shell --jars hologres-connector-spark-3.x-1.5.2-jar-with-dependencies.jar

-- For local testing, use an absolute path to load the file.
scala> :load D:/sparkselect.scala
```
- Alternatively, after the dependencies are loaded, you can paste the sample code directly into the shell to run it.
Java
You can use a development tool to import the sample code and package it using a Maven tool. For example, if the packaged JAR is named spark_select.jar, run the job with the following command.
```
-- Use an absolute path for the job JAR.
spark-submit --class SparkSelect --jars hologres-connector-spark-3.x-1.5.2-jar-with-dependencies.jar  D:\spark_select.jar
```
Python
After you run the following command, you can paste the sample code directly into the shell to run it.
```
pyspark --jars hologres-connector-spark-3.x-1.5.2-jar-with-dependencies.jar
```

Parameters

General parameters

Parameter	Default	Required	Description
username	None	Yes	The AccessKey ID of your Alibaba Cloud account. You can obtain it on the AccessKey Management page. Alternatively, the name of the user account. After you create an account, grant it the required database permissions. For more information, see Hologres permission model. Then, connect to HoloWeb and run queries to verify the permissions.
password	None	Yes	The AccessKey Secret that corresponds to your AccessKey ID. For more information, see Create an AccessKey. Alternatively, the password for the account.
table	None	Yes	The name of the Hologres table to read from or write to. Note When reading data, you can use the `read.query` parameter instead.
jdbcurl	None	Yes	The JDBC URL of the Hologres real-time data API, in the format `jdbc:postgresql://<host>:<port>/<db_name>`. To find these details, go to the Hologres console, click Instances in the left-side navigation pane, select your target instance, and find the host and port number in the Network Information section of the Instance Details page.
enable_serverless_computing	false	No	Specifies whether to use a serverless computing resource. This parameter applies only to read operations and write operations in `bulk_load` mode. For more information, see Serverless Computing user guide.
serverless_computing_query_priority	3	No	The execution priority for Serverless Computing.
statement_timeout_seconds	28800 (8 hours)	No	The query execution timeout duration, in seconds.
retry_count	3	No	The number of retries if a connection fails.
direct_connect	A direct connection is used by default if supported.	No	Specifies whether to connect directly to the Hologres FrontEnd (access node). The network throughput of the endpoint is often the bottleneck for batch data operations. By default, the connector uses a direct connection if one is available to improve throughput. Set this to `false` to disable this behavior.

Write parameters

The Hologres connector supports Spark's SaveMode parameter. For SQL, this corresponds to INSERT INTO or INSERT OVERWRITE. For a DataFrame, you can set the SaveMode to Append or Overwrite when writing data. The Overwrite mode creates a temporary table for the write operation and replaces the original table upon success. Use this mode only when necessary.

Parameter	Previous name	Default	Required	Description
write.mode	copy_write_mode	auto	No	The write mode. For a comparison of write modes, see Batch write modes. Valid values: `auto` (Default): The connector automatically selects the optimal mode based on the Hologres version and the destination table's metadata. The selection logic is as follows: If the Hologres instance is v2.2.25 or later and the table has a primary key, `bulk_load_on_conflict` mode is used. If the Hologres instance is v2.1.0 or later and the table does not have a primary key, `bulk_load` mode is used. If the Hologres instance is v1.3 or later, `stream` mode is used. In other cases, `insert` mode is used. `stream`: Uses Fixed Plan to accelerate SQL execution. In a Fixed Plan, `COPY` is a feature introduced in Hologres v1.3. Compared to the `INSERT` method, `COPY` provides higher throughput (due to its streaming nature), lower data latency, and reduced client memory consumption (as it does not batch data). Note Requires Hologres connector v1.3.0 or later and Hologres v1.3.34 or later. `bulk_load`: Uses batch `COPY`. Compared to the streaming `COPY` in a Fixed Plan, batch `COPY` imposes a lower load on the Hologres instance under high RPS conditions but only supports writing to tables without a primary key. Note Requires Hologres connector v1.4.2 or later and Hologres v2.1.0 or later. `bulk_load_on_conflict`: Uses batch `COPY` to write to tables with a primary key and can handle duplicate primary keys. By default, batch data import into a Hologres table with a primary key triggers a table lock, limiting concurrent write operations from multiple connections. The connector can redistribute data by the destination table's distribution key, allowing each Spark task to write to a single shard. This reduces the table lock to a shard-level lock, enabling concurrent writes and improving write performance. Since each connection maintains data for only a few shards, this optimization also significantly reduces the number of small files and lowers Hologres memory usage. Tests show that repartitioning data before a concurrent write can reduce the system load by about 67% compared to the `stream` mode. Note Requires Hologres connector v1.4.2 or later and Hologres v2.2.25 or later. `insert`: Writes data using the `INSERT` method. `stage`: Uses a Stage table for batch writes. This mode imports data in near-real time through a temporary storage stage and then loads it into the destination table using an `INSERT FROM SELECT` statement. In batch scenarios, this mode offers better performance and lower system load than other modes. It also supports using a serverless computing resource. Both `INSERT INTO` and `INSERT OVERWRITE` operations are supported. Requires Hologres v4.1.0 or later and connector v1.6.1 or later.
write.copy.max_buffer_size	max_cell_buffer_size	52428800 (50 MB)	No	The maximum size of the local buffer when writing in `COPY` mode. You do not usually need to adjust this value. However, you can increase this value if a buffer overflow occurs when writing large fields, such as extra-long strings.
write.copy.dirty_data_check	copy_write_dirty_data_check	false	No	Specifies whether to check for dirty data. If enabled, this feature can pinpoint the exact row that failed to write. However, this affects write performance. Keep this feature disabled except for troubleshooting.
write.on_conflict_action	INSERT_OR_REPLACE	INSERT_OR_REPLACE	No	The action to take when a write operation encounters a primary key conflict in the destination table. INSERT_OR_IGNORE: If a primary key conflict occurs, the data is not written. INSERT_OR_UPDATE: If a primary key conflict occurs, the specified columns are updated. INSERT_OR_REPLACE: If a primary key conflict occurs, all columns are updated.
write.stage.compression	None	false	No	Takes effect when `write.mode` is set to `stage`. Setting this to `true` enables compression during stage writes. Requires Hologres v4.2.8 or later and connector v1.6.2 or later.

The following parameters take effect only when write.mode is set to insert.

Parameter	Previous name	Default	Required	Description
write.insert.dynamic_partition	dynamic_partition	false	No	When `write.mode` is `insert`, setting this to `true` automatically creates non-existent partitions when writing to a parent partitioned table.
write.insert.batch_size	write_batch_size	512	No	The maximum batch size for each write thread. A batch commit is triggered when the number of `Put` operations reaches this value.
write.insert.batch_byte_size	write_batch_byte_size	2097152 (2 MB)	No	The maximum batch size in bytes for each write thread. The default value is 2 MB. A batch commit is triggered when the byte size of the `Put` data reaches this value.
write.insert.max_interval_ms	write_max_interval_ms	10000	No	A batch commit is triggered if the time elapsed since the last commit exceeds this value.
write.insert.thread_size	write_thread_size	1	No	The number of concurrent write threads. Each thread uses one database connection.
write.rps_limit	None	-1	No	The rate limit for writes per task, in rows per second (RPS). The default value of -1 indicates no limit.

Read parameters

Parameter	Previous name (v1.5.0 and earlier)	Default	Required	Description
read.mode	bulk_read	auto	No	The read mode. Valid values: `auto` (Default): The connector automatically selects the optimal mode based on the Hologres version and table metadata. The selection logic is as follows: If the fields to be read include the `JSONB` data type, `select` mode is used. If the instance is v3.0.24 or later, bulk_read_compressed mode is used. In other cases, bulk_read mode is used. `bulk_read`: Uses the `COPY OUT` method to read data in `arrow` format, which is several times faster than the `select` mode. Reading the `JSONB` type from Hologres is not currently supported. `bulk_read_compressed`: Reads compressed data in `Arrow` format using the COPY OUT method, which saves approximately 45% of bandwidth compared to reading uncompressed data. `select`: Uses standard `SELECT` statements to read data.
read.max_task_count	max_partition_count	80	No	Specifies the maximum number of concurrent tasks for reading data. The connector splits the table into multiple partitions, and each partition is processed by a single Spark task. If the table's shard count is less than this value, the number of partitions is capped at the shard count.
read.copy.max_buffer_size	/	52428800 (50 MB)	No	The maximum size of the local buffer when reading in `COPY` mode. If an exception occurs due to a large field size, increase this value.
read.push_down_predicate	push_down_predicate	true	No	Specifies whether to enable predicate pushdown. When enabled, operations such as filter conditions and column pruning are pushed down to the data source.
read.push_down_limit	push_down_limit	true	No	Specifies whether to enable `LIMIT` pushdown.
read.select.batch_size	scan_batch_size	256	No	Takes effect when `read.mode` is set to `select`. Specifies the number of rows to fetch in a single scan operation when reading from Hologres.
read.select.timeout_seconds	scan_timeout_seconds	60	No	Takes effect when `read.mode` is set to `select`. Specifies the timeout duration for a scan operation when reading from Hologres.
read.query	query	None	No	Use the provided `query` to read from Hologres. This parameter cannot be used with the `table` parameter. Note When you read data by using the `query` method, you can use only a single task. Predicate pushdown is not supported. When you read data by using the `table` method, the read operation is divided into multiple Tasks that run in parallel based on the ShardCount of the Hologres table.
read.split.strategy	None	auto	No	Defines the strategy for splitting table data into multiple tasks for parallel reads. Valid values: `auto` (default): The connector automatically selects the optimal strategy. `shard`: Shards the data based on the shards in Hologres, and each task reads a portion of the shards. This is suitable for tables with a clear shard distribution. `range`: Shards data based on the value range of a specified column. This setting requires the `read.split.column`, `read.split.lower_bound`, `read.split.upper_bound`, and `read.split.num` parameters. `partition` shards data based on the values of the partition column. This option must be used with the `read.split.column` and `read.split.num` parameters. It applies to partitioned tables or partitioned views. Note Supported in connector v1.6.1 and later.
read.split.column	None	None	No	The name of the split column. This parameter is required when `read.split.strategy` is `range` or `partition`.
read.split.lower_bound	None	None	No	The split lower bound is required when `read.split.strategy` is set to `range`. It specifies the lower bound value for range splitting.
read.split.upper_bound	None	None	No	The split upper bound specifies the upper bound for a range-based shard. This parameter is required when `read.split.strategy` is set to `range`.
read.split.num	None	None	No	The number of splits. This parameter is required when `read.split.strategy` is `range` or `partition`. It specifies the number of tasks into which the data is divided for reading.

Data type mapping

Spark type	Hologres type
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[]

Connection count calculation

Hologres-Connector-Spark uses a certain number of JDBC connections for read and write operations. The connection count depends on the following factors:

Spark parallelism, the number of concurrent tasks running during job execution, which are visible in the Spark UI.
The number of connections used per task:
- When writing in COPY mode, each task uses one JDBC connection.
- When writing in INSERT mode, each task uses write_thread_size JDBC connections.
- When reading data, each task uses one JDBC connection.
Other operations: A job may briefly use one connection for tasks like schema acquisition when it starts.

You can calculate the total connection count for a job with the following formulas:

Item	Connections
Querying metadata from the catalog	1
Reading data	parallelism * 1 + 1
Writing in COPY mode	parallelism * 1 + 1
Writing in INSERT mode	parallelism * write_thread_size + 1

This calculation assumes that Spark's capacity for concurrent tasks exceeds the number of tasks the job generates.

Spark's capacity for concurrent tasks depends on user-configured parameters, such as spark.executor.instances, and Hadoop's file block splitting policy. For more information, see Apache Hadoop.

Spark Connector release notes

Version	Release date	New features	Bug fixes
1.6.1	2026-02	Added the stage write mode, which uses a temporary stage table to improve write performance. This feature requires Hologres v4.1.0 or later. Added support for `INSERT INTO` and `INSERT OVERWRITE` operations in stage mode. Optimized the temporary table cleanup logic for `overwrite` operations. Added the `write.rps_limit` parameter to control the write rate per task.	Fixed a forced sync replay issue that occurred when executing DDL during an `overwrite` operation.
1.6.0	2025-12	Added the `read.split.strategy` parameter to support three read splitting strategies: `shard`, `range`, and `partition`. Added support for the `JSONB` data type in Arrow-based reads. Added the `write.insert.ignore_null_when_update` parameter to ignore NULL values during `UPDATE` operations. Added support for JaCoCo test coverage reporting.	Fixed an error when writing `date` values earlier than 1970.
1.5.6	2025-11	Added support for the AKV4 authentication method. Added the `write.copy.disable_right_join` parameter to optimize `COPY` write performance.	Optimized the column type checking logic for catalog writes.
1.5.5	2025-10	Added support for writing low-precision data to high-precision data types. Added prefixes such as `appname` and `taskid` to logs to simplify troubleshooting. Enabled Arrow LZ4 compression for `COPY OUT` operations.	Optimized parameter formatting and validation.
1.5.4	2025-09	Added the `remove_u0000` parameter to automatically remove `U+0000` characters from text columns. Refactored the Hologres Catalog to map namespaces to schemas. Added the `RepartitionUtil` utility class for data sharding.	Refactored documentation and updated links.
1.5.2	2025-08	Introduced Hologres Catalog, which enables reading from and writing to Hologres by using external tables. Added support for predicate pushdown and limit pushdown. Enabled the `bulk_read` batch mode for read operations, improving performance by an order of magnitude.	Optimized read performance.
1.5.0	2025-06	Added support for predicate pushdown, limit pushdown, and column pruning. Enabled reading data by using `SELECT` queries. Enabled batch mode for read operations.	Fixed read-related issues.
1.4.2	2025-04	Introduced the `bulk_load` write mode for batch writing to tables without a primary key. Added the `bulk_load_on_conflict` mode to handle primary key conflicts in tables with a primary key.	Optimized write performance.
1.3.2	2025-02	Added support for reading data from Hologres. Enabled parallel reads by shard. Added support for automatic schema inference.	Stabilized the initial read functionality.
1.3.0	2025-01	Introduced the `stream` (fixed `COPY`) write mode. Added support for the fixed `COPY` feature in Hologres v1.3.	Optimized write throughput and latency.