GreptimeDB Apache Spark Connector

spark-connector-greptimedb is a Spark DataSource V2 connector for writing batch DataFrames and Structured Streaming micro-batches to GreptimeDB.

Compatibility

• Apache Spark 4.2.0
• Java 17 or later
• Scala 2.13 Spark distribution
• GreptimeDB Java ingester 0.15.0

Maven dependency

<dependency>
    <groupId>io.greptime</groupId>
    <artifactId>spark-connector-greptimedb</artifactId>
    <version>${connector.version}</version>
</dependency>

When submitting an application without dependency resolution, use the -shaded.jar produced by this project.

GreptimeDB table

The connector writes to an existing table; it does not create or alter tables. It does not validate the server-side schema before writing. Ensure the Spark columns and mapped types match the target table; otherwise a later bulk batch can fail after earlier batches have already been written.

CREATE TABLE cpu_metrics (
    ts TIMESTAMP(6) TIME INDEX,
    host STRING,
    usage DOUBLE,
    PRIMARY KEY (host)
);

Batch write

Dataset<Row> metrics = ...;

metrics.write()
        .format("greptimedb")
        .mode("append")
        .option("endpoints", "127.0.0.1:4001")
        .option("database", "public")
        .option("table", "cpu_metrics")
        .option("time-index", "ts")
        .option("tags", "host")
        .option("batch.max-rows", "1000")
        .save();

Structured Streaming write

StreamingQuery query = metrics.writeStream()
        .format("greptimedb")
        .outputMode("append")
        .option("checkpointLocation", "/path/to/checkpoint")
        .option("endpoints", "127.0.0.1:4001")
        .option("database", "public")
        .option("table", "cpu_metrics")
        .option("time-index", "ts")
        .option("tags", "host")
        .start();

Options

Option	Required	Default	Description
endpoints	Yes	-	Comma-separated GreptimeDB gRPC endpoints in hostname:port or IPv4:port format. IPv6 is not supported by the current ingester SDK.
table	Yes	-	Existing target table.
time-index	Yes	-	Spark TIMESTAMP or TIMESTAMP_NTZ column used as the GreptimeDB time index. Its row values must not be null.
database	No	public	Target database.
tags	No	empty	Comma-separated columns written with tag semantics. Other non-time-index columns are fields.
username	No	-	Username; must be configured together with password.
password	No	-	Password; must be configured together with username.
batch.max-rows	No	1000	Rows per ingester bulk message in each Spark task.
bulk.timeout-ms-per-message	No	SDK default	Timeout for each bulk message.
bulk.max-requests-in-flight	No	SDK default (currently 8)	Maximum asynchronous bulk messages in flight per Spark task.
bulk.allocator-init-reservation-bytes	No	SDK default	Initial Arrow allocator reservation.
bulk.allocator-max-allocation-bytes	No	SDK default	Maximum Arrow allocator allocation.

Type mapping

Spark SQL type	GreptimeDB type
BOOLEAN	BOOLEAN
TINYINT	INT8
SMALLINT	INT16
INT	INT32
BIGINT	INT64
FLOAT	FLOAT32
DOUBLE	FLOAT64
STRING, CHAR, VARCHAR	STRING
BINARY	BINARY
DATE	DATE
TIMESTAMP, TIMESTAMP_NTZ	TIMESTAMP_MICROSECOND
DECIMAL(p, s)	DECIMAL128(p, s)

Complex Spark SQL types are rejected before a job is submitted.

Delivery semantics

Writes are insert-only and at-least-once. GreptimeDB ingester writes become visible before Spark's driver-level commit callback, so failed or retried Spark tasks and retried streaming epochs can produce duplicate rows. The connector does not implement overwrite, delete, upsert, automatic DDL, or exactly-once commit.

Build and test

mvn test
mvn package

The integration test starts GreptimeDB with Testcontainers and executes a local Spark write:

mvn -Pintegration-test verify

Use a different GreptimeDB image when needed:

mvn -Pintegration-test verify -Dgreptimedb.test.image=greptime/greptimedb:latest

Build output:

• target/spark-connector-greptimedb-${version}.jar: thin Maven artifact.
• target/spark-connector-greptimedb-${version}-shaded.jar: deployment artifact containing the GreptimeDB ingester client while excluding Spark and Scala.