Hologres:Access Hologres via Spark

Write using CATALOG

-- Load the 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 ","  -- Use the absolute path for local testing
);

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

Write using 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 ","
);

-- Declare a Hologres temporary view with only the target columns
CREATE TEMPORARY VIEW hologresTable (
    c_custkey BIGINT,
    c_name    STRING,
    c_phone   STRING)
USING hologres OPTIONS (
    jdbcurl  "jdbc:postgresql://<host>:<port>/test_db",
    username "<your-username>",
    password "<your-password>",
    table    "customer_holo_table"
);

-- Write only the selected columns
INSERT INTO hologresTable SELECT c_custkey, c_name, c_phone FROM csvTable;

Scala

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

// Define the CSV schema
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 from CSV
val csvDf = spark.read.format("csv").schema(schema).option("sep", ",").load("resources/customer")

// Write to Hologres
csvDf.write
  .format("hologres")
  .option("username", "<your-username>")
  .option("password", "<your-password>")
  .option("jdbcurl",  "jdbc:postgresql://<host>:<port>/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) {
        // Define the CSV schema
        List<StructField> fields = 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(fields);

        SparkSession spark = SparkSession.builder()
            .appName("Spark Hologres Write Example")
            .master("local[*]")
            .getOrCreate();

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

        // Write to Hologres
        csvDf.write()
            .format("hologres")
            .option("username", "<your-username>")
            .option("password", "<your-password>")
            .option("jdbcurl",  "jdbc:postgresql://<host>:<port>/test_db")
            .option("table",    "customer_holo_table")
            .mode("append")
            .save();
    }
}

Add the following Maven dependency for Spark SQL:

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

# Define the CSV schema
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 from CSV
csvDf = spark.read.csv("resources/customer", header=False, schema=schema, sep=",")

# Write to Hologres
csvDf.write.format("hologres") \
    .option("username", "<your-username>") \
    .option("password", "<your-password>") \
    .option("jdbcurl",  "jdbc:postgresql://<host>:<port>/test_db") \
    .option("table",    "customer_holo_table") \
    .mode("append") \
    .save()

CREATE TEMPORARY VIEW(table)

-- Declare a Hologres table as a temporary view
CREATE TEMPORARY VIEW hologresTable
USING hologres OPTIONS (
  jdbcurl           "jdbc:postgresql://<host>:<port>/test_db",
  username          "<your-username>",
  password          "<your-password>",
  read.max_task_count "80",  -- Maximum concurrent read tasks
  table             "customer_holo_table"
);

-- Column pruning and predicate pushdown are applied automatically
SELECT c_custkey, c_name, c_phone FROM hologresTable WHERE c_custkey < 500 LIMIT 10;

CREATE TEMPORARY VIEW(query)

-- Pass a Hologres SELECT query directly
CREATE TEMPORARY VIEW hologresTable
USING hologres OPTIONS (
  jdbcurl    "jdbc:postgresql://<host>:<port>/test_db",
  username   "<your-username>",
  password   "<your-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;

Scala

val readDf = spark.read
  .format("hologres")
  .option("username",           "<your-username>")
  .option("password",           "<your-password>")
  .option("jdbcurl",            "jdbc:postgresql://<host>:<port>/test_db")
  .option("table",              "customer_holo_table")
  .option("read.max_task_count", "80")  // Maximum concurrent read tasks
  .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) {
        SparkSession spark = SparkSession.builder()
            .appName("Spark Hologres Read Example")
            .master("local[*]")
            .getOrCreate();

        Dataset<Row> readDf = spark.read()
            .format("hologres")
            .option("username",            "<your-username>")
            .option("password",            "<your-password>")
            .option("jdbcurl",             "jdbc:postgresql://<host>:<port>/test_db")
            .option("table",              "customer_holo_table")
            .option("read.max_task_count", "80")  // Maximum concurrent read tasks
            .load()
            .filter("c_custkey < 500");

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

Add the same Maven dependency as in the write example (Spark SQL spark-sql_2.13 version 3.5.4). To submit the job:

spark-submit --class SparkSelect \
  --jars hologres-connector-spark-3.x-1.5.2-jar-with-dependencies.jar \
  D:\spark_select.jar

Python

readDf = spark.read.format("hologres") \
    .option("username",            "<your-username>") \
    .option("password",            "<your-password>") \
    .option("jdbcurl",             "jdbc:postgresql://<host>:<port>/test_db") \
    .option("table",               "customer_holo_table") \
    .option("read.max_task_count", "80") \
    .load()

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