Edit

Spark-Pinot Connector

Use the Spark-Pinot connector to read data from and write data to Pinot.

Spark-pinot connector to read data from Pinot.

Detailed read model documentation is here: Spark Pinot Connector Read Model

The write model is Experimental and the documentation is here: Spark Pinot Connector Write Model

Features

  • Query realtime, offline or hybrid tables

  • Distributed, parallel scan

  • Streaming reads using gRPC (optional)

  • SQL support instead of PQL

  • Column and filter push down to optimize performance

  • Overlap between realtime and offline segments is queried exactly once for hybrid tables

  • Schema discovery

    • Dynamic inference

    • Static analysis of case class

  • Supports query options

  • HTTPS/TLS support for secure connections

Quick Start

import org.apache.spark.sql.SparkSession

val spark: SparkSession = SparkSession
      .builder()
      .appName("spark-pinot-connector-test")
      .master("local")
      .getOrCreate()

import spark.implicits._

val data = spark.read
  .format("pinot")
  .option("table", "airlineStats")
  .option("tableType", "offline")
  .load()
  .filter($"DestStateName" === "Florida")

data.show(100)

Security Configuration

This feature is supported after 1.4.0 release, please use current master or wait for 1.5.0

You can secure both HTTP and gRPC using a unified switch or explicit flags.

  • Unified: set secureMode=true to enable HTTPS and gRPC TLS together (recommended)

  • Explicit: set useHttps for REST and grpc.use-plain-text=false for gRPC

Quick examples

// Unified secure mode (enables HTTPS + gRPC TLS by default)
val data = spark.read
  .format("pinot")
  .option("table", "airlineStats")
  .option("tableType", "offline")
  .option("secureMode", "true")
  .load()

// Explicit HTTPS only (gRPC remains plaintext by default)
val data = spark.read
  .format("pinot")
  .option("table", "airlineStats")
  .option("tableType", "offline")
  .option("useHttps", "true")
  .load()

// Explicit gRPC TLS only (REST remains HTTP by default)
val data = spark.read
  .format("pinot")
  .option("table", "airlineStats")
  .option("tableType", "offline")
  .option("grpc.use-plain-text", "false")
  .load()

HTTPS Configuration

When HTTPS is enabled (either via secureMode=true or useHttps=true), you can configure keystore/truststore as needed:

val data = spark.read
  .format("pinot")
  .option("table", "airlineStats")
  .option("tableType", "offline")
  .option("useHttps", "true")
  .option("keystorePath", "/path/to/keystore.jks")
  .option("keystorePassword", "keystorePassword")
  .option("truststorePath", "/path/to/truststore.jks")
  .option("truststorePassword", "truststorePassword")
  .load()

HTTPS Configuration Options

Option
Description
Required
Default

secureMode

Unified switch to enable HTTPS and gRPC TLS

No

false

useHttps

Enable HTTPS connections (overrides secureMode for REST)

No

false

keystorePath

Path to client keystore file (JKS format)

No

None

keystorePassword

Password for the keystore

No

None

truststorePath

Path to truststore file (JKS format)

No

None

truststorePassword

Password for the truststore

No

None

Note: If no truststore is provided when HTTPS is enabled, the connector will trust all certificates (not recommended for production use).

Authentication Support

This feature is supported after 1.4.0 release, please use current master or wait for 1.5.0

The connector supports custom authentication headers for secure access to Pinot clusters:

// Using Bearer token authentication
val data = spark.read
  .format("pinot")
  .option("table", "airlineStats")
  .option("tableType", "offline")
  .option("authToken", "my-jwt-token")  // Automatically adds "Authorization: Bearer my-jwt-token"
  .load()

// Using custom authentication header
val data = spark.read
  .format("pinot")
  .option("table", "airlineStats")
  .option("tableType", "offline")
  .option("authHeader", "Authorization")
  .option("authToken", "Bearer my-custom-token")
  .load()

// Using API key authentication
val data = spark.read
  .format("pinot")
  .option("table", "airlineStats")
  .option("tableType", "offline")
  .option("authHeader", "X-API-Key")
  .option("authToken", "my-api-key")
  .load()

Authentication Configuration Options

This feature is supported after 1.4.0 release, please use current master or wait for 1.5.0

Option
Description
Required
Default

authHeader

Custom authentication header name

No

Authorization (when authToken is provided)

authToken

Authentication token/value

No

None

Note: If only authToken is provided without authHeader, the connector will automatically use Authorization: Bearer <token>.

Pinot Proxy Support

This feature is supported after 1.4.0 release, please use current master or wait for 1.5.0

The connector supports Pinot Proxy for secure cluster access where the proxy is the only exposed endpoint. When proxy is enabled, all HTTP requests to controllers/brokers and gRPC requests to servers are routed through the proxy.

Proxy Configuration Examples

// Basic proxy configuration
val data = spark.read
  .format("pinot")
  .option("table", "airlineStats")
  .option("tableType", "offline")
  .option("controller", "pinot-proxy:8080")  // Proxy endpoint
  .option("proxy.enabled", "true")
  .load()

// Proxy with authentication
val data = spark.read
  .format("pinot")
  .option("table", "airlineStats")
  .option("tableType", "offline")
  .option("controller", "pinot-proxy:8080")
  .option("proxy.enabled", "true")
  .option("authToken", "my-proxy-token")
  .load()

// Proxy with gRPC configuration
val data = spark.read
  .format("pinot")
  .option("table", "airlineStats")
  .option("tableType", "offline")
  .option("controller", "pinot-proxy:8080")
  .option("proxy.enabled", "true")
  .option("grpc.proxy-uri", "pinot-proxy:8094")  // gRPC proxy endpoint
  .load()

Proxy Configuration Options

Option
Description
Required
Default

proxy.enabled

Use Pinot Proxy for controller and broker requests

No

false

Note: When proxy is enabled, the connector adds FORWARD_HOST and FORWARD_PORT headers to route requests to the actual Pinot services.

gRPC Configuration

This feature is supported after 1.4.0 release, please use current master or wait for 1.5.0

The connector supports comprehensive gRPC configuration for secure and optimized communication with Pinot servers.

gRPC Configuration Examples

// Basic gRPC configuration
val data = spark.read
  .format("pinot")
  .option("table", "airlineStats")
  .option("tableType", "offline")
  .option("grpc.port", "8091")
  .option("grpc.max-inbound-message-size", "256000000")  // 256MB
  .load()

// gRPC with TLS (explicit)
val data = spark.read
  .format("pinot")
  .option("table", "airlineStats")
  .option("tableType", "offline")
  .option("grpc.use-plain-text", "false")
  .option("grpc.tls.keystore-path", "/path/to/grpc-keystore.jks")
  .option("grpc.tls.keystore-password", "keystore-password")
  .option("grpc.tls.truststore-path", "/path/to/grpc-truststore.jks")
  .option("grpc.tls.truststore-password", "truststore-password")
  .load()

// gRPC with proxy
val data = spark.read
  .format("pinot")
  .option("table", "airlineStats")
  .option("tableType", "offline")
  .option("proxy.enabled", "true")
  .option("grpc.proxy-uri", "pinot-proxy:8094")
  .load()

gRPC Configuration Options

Option
Description
Required
Default

grpc.port

Pinot gRPC port

No

8090

grpc.max-inbound-message-size

Max inbound message bytes when init gRPC client

No

128MB

grpc.use-plain-text

Use plain text for gRPC communication (overrides secureMode for gRPC)

No

true

grpc.tls.keystore-type

TLS keystore type for gRPC connection

No

JKS

grpc.tls.keystore-path

TLS keystore file location for gRPC connection

No

None

grpc.tls.keystore-password

TLS keystore password

No

None

grpc.tls.truststore-type

TLS truststore type for gRPC connection

No

JKS

grpc.tls.truststore-path

TLS truststore file location for gRPC connection

No

None

grpc.tls.truststore-password

TLS truststore password

No

None

grpc.tls.ssl-provider

SSL provider

No

JDK

grpc.proxy-uri

Pinot Rest Proxy gRPC endpoint URI

No

None

Note: When using gRPC with proxy, the connector automatically adds FORWARD_HOST and FORWARD_PORT metadata headers for proper request routing.

Example run with spark-shell

There are examples under https://github.com/apache/pinot/tree/master/pinot-connectors/pinot-spark-3-connector/examples .

Prerequisites

  • Apache Spark 3.x installed and spark-shell available in your PATH.

  • Setup PINOT_HOME env variable:

    export PINOT_HOME=/path/to/pinot

  • The Pinot Spark 3 Connector shaded JAR built and available at:

    $PINOT_HOME/pinot-connectors/pinot-spark-3-connector/target/pinot-spark-3-connector-*-shaded.jar

  • Example Scala script located at:

    $PINOT_HOME/pinot-connectors/pinot-spark-3-connector/examples/read_pinot_from_proxy_with_auth_token.scala

Scala Script to read data from Pinot Proxy

read_pinot_from_proxy_with_auth_token.scala
import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder().appName("read-pinot-airlineStats").master("local[*]").getOrCreate()

val df = spark.read.
  format("org.apache.pinot.connector.spark.v3.datasource.PinotDataSource").
  option("table", "myTable").
  option("tableType", "offline").
  option("controller", "pinot-proxy:8080").
  option("secureMode", "true").
  option("authToken", "st-xxxxxxx").
  option("proxy.enabled", "true").
  option("grpc.proxy-uri", "pinot-proxy:8094").
  option("useGrpcServer", "true").
  load()

println("Schema:")
df.printSchema()

println("Sample rows:")
df.show(10, truncate = false)

println(s"Total rows: ${df.count()}")

spark.stop()

Run with spark-shell

Launch the example in spark-shell with the following command:

spark-shell 
    --master 'local[*]' \
    --name read-pinot \
    --jars "$PINOT_HOME/pinot-connectors/pinot-spark-3-connector/target/pinot-spark-3-connector-*-shaded.jar" < "$PINOT_HOME/pinot-connectors/pinot-spark-3-connector/examples/read_pinot_from_proxy_with_auth_token.scala"

Sample output

spark-shell --master 'local[*]' --name read-pinot --jars "$PINOT_HOME/pinot-connectors/pinot-spark-3-connector/target/pinot-spark-3-connector-*-shaded.jar" < "$PINOT_HOME/pinot-connectors/pinot-spark-3-connector/examples/read_pinot_from_proxy_with_auth_token.scala"

25/09/04 07:59:29 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
Spark context Web UI available at http://xiang-mac-home.wyvern-sun.ts.net:4040
Spark context available as 'sc' (master = local[*], app id = local-1756997971428).
Spark session available as 'spark'.
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 3.5.1
      /_/

Using Scala version 2.12.18 (OpenJDK 64-Bit Server VM, Java 17.0.15)
Type in expressions to have them evaluated.
Type :help for more information.

scala> import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.SparkSession

scala>

scala> val spark = SparkSession.builder().appName("read-pinot-table").master("local[*]").getOrCreate()
25/09/04 07:59:35 WARN SparkSession: Using an existing Spark session; only runtime SQL configurations will take effect.
spark: org.apache.spark.sql.SparkSession = org.apache.spark.sql.SparkSession@c377641

scala>

scala> val df = spark.read.
     |   format("org.apache.pinot.connector.spark.v3.datasource.PinotDataSource").
     |   option("table", "api_gateway_agg_monthly").
     |   option("tableType", "REALTIME").
     |   option("controller", "pinot.xxx.yyy.startree.cloud").
     |   option("broker", "broker.pinot.xxx.yyy.startree.cloud").
     |   option("secureMode", "true").
     |   option("authToken", "st-xxx-yyy").
     |   option("proxy.enabled", "true").
     |   option("grpc.proxy-uri", "proxy-grpc.pinot.xxx.yyy.startree.cloud").
     |   option("useGrpcServer", "true").
     |   load()
25/09/04 07:59:35 WARN HttpUtils: No truststore configured, trusting all certificates (not recommended for production)
df: org.apache.spark.sql.DataFrame = [api_calls_count: bigint, developer_account_id: string ... 1 more field]

scala>

scala> println("Schema:")
Schema:

scala> df.printSchema()
root
 |-- api_calls_count: long (nullable = true)
 |-- developer_account_id: string (nullable = true)
 |-- monthsSinceEpoch: long (nullable = true)


scala>

scala> println("Sample rows:")
Sample rows:

scala> df.show(10, truncate = false)
25/09/04 07:59:39 WARN HttpUtils: No truststore configured, trusting all certificates (not recommended for production)
+---------------+------------------------------------+----------------+
|api_calls_count|developer_account_id                |monthsSinceEpoch|
+---------------+------------------------------------+----------------+
|276            |000e2e63-12ef-e353-af76-6fe98d2e8747|1748736000000   |
|287            |00101768-41ba-0f01-36da-2dfee58f4703|1748736000000   |
|287            |00121c0f-3825-364e-18e4-dac8dd0ea8d1|1748736000000   |
|290            |00124c8f-2a3d-fe73-3090-482c3757af09|1748736000000   |
|299            |00128fa3-a508-5e39-0ab9-c3f2039a878a|1748736000000   |
|275            |0014e7b0-9d48-2ef9-9111-e830e74f32c4|1748736000000   |
|293            |0017b314-b083-9240-0117-515519c22fd4|1748736000000   |
|306            |0026de58-e04a-ec08-fe94-beb363324f30|1748736000000   |
|297            |002dff23-dbd7-8ef3-a55f-f225523519b4|1748736000000   |
|277            |002f40b0-409c-b3e8-bb69-049b3e321589|1748736000000   |
+---------------+------------------------------------+----------------+
only showing top 10 rows


scala>

scala> println(s"Total rows: ${df.count()}")
25/09/04 08:00:38 WARN HttpUtils: No truststore configured, trusting all certificates (not recommended for production)
Total rows: 60000

scala>

scala> spark.stop()

scala> :quit

Example run with spark-submit

You can run the examples locally (e.g. using your IDE) in a standalone mode by starting a local Pinot cluster. See: https://docs.pinot.apache.org/basics/getting-started/running-pinot-locally

You can also run the tests in cluster mode using following command:

export SPARK_CLUSTER=<YOUR_YARN_OR_SPARK_CLUSTER>

# Edit the ExampleSparkPinotConnectorTest to get rid of `.master("local")` and rebuild the jar before running this command
spark-submit \
    --class org.apache.pinot.connector.spark.v3.datasource.ExampleSparkPinotConnectorTest \
    --jars ./target/pinot-spark-3-connector-1.3.0-shaded.jar \
    --master $SPARK_CLUSTER \
    --deploy-mode cluster \
  ./target/pinot-spark-3-connector-1.3.0-tests.jar

This example demonstrates how to use the Pinot Spark 3 Connector to read data from a Pinot cluster via a proxy with authentication token support.

Security Best Practices

This feature is supported after 1.4.0 release, please use current master or wait for 1.5.0

Production HTTPS Configuration

  • Always use HTTPS in production environments

  • Store certificates in secure locations with appropriate file permissions

  • Use proper certificate validation with valid truststore

  • Rotate certificates regularly

Production Authentication

  • Use service accounts with minimal required permissions

  • Store authentication tokens securely (environment variables, secret management systems)

  • Implement token rotation policies

  • Monitor authentication failures

Production gRPC Configuration

  • Enable TLS for gRPC communication in production

  • Use certificate-based authentication when possible

  • Configure appropriate message size limits based on your data

  • Use connection pooling for high-throughput scenarios

Future Works

  • Add integration tests for read operation

  • Add write support(pinot segment write logic will be changed in later versions of pinot)

PreviousPrestoNextSpark Pinot Connector Read Model

Last updated

Was this helpful?