httpds-spark

Read files from HTTP APIs with Apache Spark, in batch or as a micro-batch stream. The connector uses Spark DataSource V2. To connect your file source, implement a small client interface that lists and downloads files.

Features

• Batch and streaming — read existing files or keep polling for new ones.
• Custom clients — connect your HTTP API by implementing one Scala trait.
• Background downloads — optionally download files ahead of processing.
• Spark UI metrics — track downloaded files, bytes, and download speed.
• Streaming start positions — start from the first file, new arrivals, or a named file.

Table of Contents

• Quick Start
• Installation
• Configuration
• Usage
  • Batch Mode
  • Streaming Mode
• Output Schema
• Implementing a Custom Client
• Architecture
• Metrics
• Building & Testing
• Contributing
• Maintainers
• License

Quick Start

This example assumes you have installed the connector and a client registered as my-client. Replace it with your client’s name; see Implementing a Custom Client to create one.

val df = spark.read
  .format("httpds")
  .option("remoteClient", "my-client")
  .option("uri", "https://files.example.com/api/v1")
  .option("partitions", "8")
  .load()

df.show()

Installation

Prerequisites

Requirement	Version
Java	11+
Scala	2.12
Apache Spark	3.5.x
sbt	1.12+

Build the JAR

sbt package

The JAR is written to target/scala-2.12/. Add it to your Spark application's classpath:

spark-submit --jars httpds-spark_2.12-1.0.0-SNAPSHOT.jar ...

sbt dependency (local publish)

sbt publishLocal

Then add to your build.sbt:

libraryDependencies += "com.amadeus.spark" %% "httpds-spark" % "1.0.0-SNAPSHOT"

Spark and SLF4J must be available in your Spark cluster; they are declared as Provided dependencies.

Configuration

Set options with .option(key, value) on spark.read or spark.readStream. They are checked before execution.

Counts must be positive, except maxRetries, which can be zero. Durations must be finite: connectTimeout must be at least 1ms, and readTimeout must be greater than zero. pollInterval and retryDelay can be zero to poll or retry immediately.

Required

Option	Description
uri / path	Base URL of the remote server (e.g. https://files.example.com/api/v1)
remoteClient	Short name or fully qualified class name of the client implementation

HTTP

The connector passes these settings to your client and applies connectTimeout to the async HTTP clients it creates. Your client must implement the other settings; check its documentation for support.

Option	Default	Description
connectTimeout	30s	HTTP connection timeout (ms, s, m suffixes supported); connectionTimeout is an alias
readTimeout	60s	HTTP read timeout, applied by your client
maxRetries	3	Maximum retries, implemented by your client
retryDelay	1s	Delay between retries, managed by your client
enableSsl	false	Client-specific TLS setting; does not enable HTTPS by itself
trustStorePath	—	Trust-store path, if supported by your client
trustStorePassword	—	Trust-store password, if supported by your client

The connector’s HTTP clients use the JVM’s default TLS settings and the URL’s http:// or https:// scheme. enableSsl does not change the URL to HTTPS. Custom trust stores require client support: setting trustStorePath and trustStorePassword alone does not configure the shared HTTP clients.

Authentication

Option	Default	Description
apiKey	—	Credential for your client
apiSecret	—	Credential for your client

Your client decides how to send these credentials, such as through basic auth, bearer tokens, or custom headers. The connector does not send them automatically.

Parallelism

Option	Default	Description
partitions	4	Number of Spark partitions for parallel download

Streaming

Option	Default	Description
pollInterval	2s	How often to poll the file index for new files
maxFilesPerTrigger	100	Maximum files to process in a single micro-batch
startOffset	earliest	earliest, latest, or file:<filename>

Async Downloads

Option	Default	Description
asyncDownloads	false	Download files ahead of processing
asyncPrefetchSize	20	Number of files to download ahead
asyncDownloadThreads	4	Number of background download threads
asyncListFiles	false	List files asynchronously in the background

Tasks with the same asyncDownloadThreads and connectTimeout share async resources. Each different pair creates a separate pool that stays open until the executor stops, so avoid using many combinations in a long-running executor.

Older option names still work, but the current names take priority:

• Threads: asyncDownloadThreads, then spark.remoteFile.asyncDownload.threads.
• Connection timeout: connectTimeout, then connectionTimeout, then spark.remoteFile.connectionTimeout (milliseconds).

All readers default to 4 threads and 30 seconds.

Other

Option	Default	Description
sessionId	—	Optional session identifier (useful for test isolation)

Usage

Batch Mode

This example assumes my-client supports the authentication and TLS options shown.

val df = spark.read
  .format("httpds")
  .option("remoteClient", "my-client")
  .option("uri", "https://files.example.com/api/v1")
  .option("apiKey", sys.env("API_KEY"))
  .option("apiSecret", sys.env("API_SECRET"))
  .option("partitions", "8")
  .option("readTimeout", "30s")
  .option("enableSsl", "true")
  .load()

df.printSchema()
df.show()

Streaming Mode

This example requires my-client to implement authentication and AsyncRemoteFileClient.

val stream = spark.readStream
  .format("httpds")
  .option("remoteClient", "my-client")
  .option("uri", "https://files.example.com/api/v1")
  .option("apiKey", sys.env("API_KEY"))
  .option("apiSecret", sys.env("API_SECRET"))
  .option("pollInterval", "10s")
  .option("maxFilesPerTrigger", "50")
  .option("startOffset", "latest")
  .option("asyncDownloads", "true")
  .option("asyncPrefetchSize", "10")
  .load()

stream.writeStream
  .format("parquet")
  .option("checkpointLocation", "/checkpoints/my-source")
  .start("/data/my-source")

Start offset values

Value	Behaviour
earliest	Process all files from the beginning
latest	Skip existing files; only process new arrivals
file:<filename>	Start from (and including) the specified filename

Output Schema

Each row has the following fields:

root
 |-- sourceFile: struct
 |    |-- name: string            — filename
 |    |-- fetchedAt: timestamp    — when the file was discovered
 |    |-- fileSize: long          — Content-Length from HTTP response
 |    |-- lastModified: timestamp — Last-Modified HTTP header
 |    |-- etag: string            — ETag HTTP header
 |    |-- requestId: string       — Request ID HTTP header
 |-- logMetadata: string          — parsed metadata from file content (if any)
 |-- logContent: binary           — file content ready for processing
 |-- rawFileBinary: binary        — full raw file bytes (fallback when parsing fails)
 |-- downloadTimestamp: timestamp  — when the file was downloaded
 |-- error: string                — error message if download/parsing failed

Implementing a Custom Client

Implement RemoteFileClient to list and download files, then register it with Java ServiceLoader.

Your client handles HTTP requests, authentication, request timeouts, retries, and response parsing. Document the options it supports and reject unsupported security settings. This also applies to async clients using the connector’s shared HTTP resources.

1. Implement the trait

package com.example.myclient

import com.amadeus.spark.datasource.remote.client._
import com.amadeus.spark.datasource.remote.conf.RemoteFileDataSourceOptions

class MyRemoteClient extends RemoteFileClient {

  private var options: RemoteFileDataSourceOptions = _

  override def init(opts: RemoteFileDataSourceOptions): Unit =
    this.options = opts

  override def shortName(): String = "my-client"

  override def listLogFiles(): Seq[RemoteFile] = {
    // Call your HTTP endpoint and return the list of available files
    ???
  }

  override def downloadLogFileWithResult(filename: String): DownloadResult = {
    // Download the file and return DownloadSuccess or DownloadFailure
    ???
  }

  override def close(): Unit = {
    // Release any resources (HTTP client, etc.)
  }
}

Implement AsyncRemoteFileClient to support background downloads, or FileSorter to control file order.

2. Register via ServiceLoader

Create the file:

src/main/resources/META-INF/services/com.amadeus.spark.datasource.remote.client.RemoteFileClientRegister

With the fully qualified class name:

com.example.myclient.MyRemoteClient

3. Use it

spark.read
  .format("httpds")
  .option("remoteClient", "my-client")   // matches shortName()
  .option("uri", "https://my-server.example.com")
  .load()

Architecture

┌──────────────────────────────────────────────────────────────────┐
│           com.amadeus.spark.datasource.remote  (framework)       │
│                                                                  │
│  ┌────────────────────┐     ┌──────────────────────────────────┐ │
│  │  RemoteFileBatch   │     │  RemoteMicroBatchStream          │ │
│  │  (batch read)      │     │  (streaming read)                │ │
│  └────────┬───────────┘     └───────────────┬──────────────────┘ │
│           │                                 │                    │
│  ┌────────▼─────────────────────────────────▼──────────────────┐ │
│  │  RemoteFilePartitionReader / AsyncRemoteFilePartitionReader │ │
│  └────────────────────────────┬────────────────────────────────┘ │
│                               │                                  │
│              RemoteFileClient (trait) ◄── ClientRegistry         │
│                     (ServiceLoader discovery)                    │
└───────────────────────────────┼──────────────────────────────────┘
                                │
                                │
                   ┌────────────▼─────────┐
                   │  Your custom client: │
                   │  MyRemoteClient      │
                   └──────────────────────┘

Key Components

Component	Description
RemoteFileTableProvider	Entry point; registered as the httpds format
RemoteFileBatch	Plans partitions and distributes files round-robin
RemoteMicroBatchStream	Tracks offsets and polls for new files
RemoteFilePartitionReader	Downloads files and returns rows for a partition
AsyncRemoteFilePartitionReader	Downloads files ahead of processing for a partition
ClientRegistry	Discovers client implementations via Java ServiceLoader
RemoteFileDataSourceOptions	Reads and checks options, and supplies defaults

Metrics

The Spark UI shows these metrics for each task:

Metric	Description
filesDownloaded	Number of files downloaded per task
bytesDownloaded	Total bytes downloaded per task
downloadDurationMs	Total download time in milliseconds
bytesThroughput	Bytes per second
recordsThroughput	Records per second

Job metrics also report minimum, maximum, and average values.

Building & Testing

Build

sbt compile    # Compile the project
sbt package    # Package the JAR

Test

sbt test       # Run all unit tests

Tests use ScalaTest and WireMock — no external services required.

Dependencies

Library	Version
Apache Spark	3.5.0 (provided)
Scala	2.12.18
sttp client4	4.0.13
ScalaTest	3.2.17
WireMock	3.13.2

Contributing

Contributions are welcome! Please read the Contributing Guide and Code of Conduct before submitting a pull request.

Maintainers

• Simone DE SANTIS (@sdeswork)
• Guillaume LECLERC (@guleclerc)

See .github/CODEOWNERS for reviewers, and SECURITY.md to report vulnerabilities.

License

This project is licensed under the Apache License 2.0.