Ingest from a custom data source¶

Some sources have no built-in Spark connector: a REST API, an internal protocol, a queue only your platform speaks. For those you write a PySpark DataSource class — and then you have to wire it into the pipeline: register it on the session, read it with the right format string, and make it survive the trip to the executors.

You could hand-write that wiring: a spark.dataSource.register(...) call, a spark.readStream.format(...).load() inside a @dp.temporary_view, and — the line everyone forgets — a cloudpickle registration so the class serializes to the executors. Or you declare a type: custom_datasource load, point it at your module and class, and let Lakehouse Plumber write all of that. That is the idea on every page: declare your ETL, don’t hand-write it.

Let’s ingest live currency exchange rates from a REST API through a DataSource you supply, and land them in a bronze streaming table.

Before you start¶

You need Lakehouse Plumber installed and a project to work in. You also need the DataSource implementation itself — writing one is a PySpark skill, not a Lakehouse Plumber one, so this guide brings a finished class and focuses on wiring it in. If the source you need already has a native load type (cloudfiles, delta, jdbc, kafka), reach for that instead; a custom data source is for the systems Spark cannot read on its own.

Bring the DataSource class¶

Put the PySpark DataSource under data_sources/ at the project root. This one streams exchange rates from an HTTP endpoint:

data_sources/exchange_rates_source.py¶

"""A PySpark DataSource that streams currency exchange rates from a REST API.

Spark ships no connector for an arbitrary HTTP endpoint, so this implements the
PySpark ``DataSource`` interface directly. The streaming reader fetches one
snapshot of rates per micro-batch and lets Spark manage the offsets.
"""

from typing import Iterator, Tuple

from pyspark.sql.datasource import (
    DataSource,
    DataSourceStreamReader,
    InputPartition,
)
from pyspark.sql.types import StructType


class ExchangeRatesDataSource(DataSource):
    """Live currency exchange rates from a REST API."""

    @classmethod
    def name(cls):
        return "exchange_rates"

    def schema(self):
        return (
            "base_currency string, "
            "quote_currency string, "
            "rate double, "
            "fetched_at timestamp"
        )

    def streamReader(self, schema: StructType):
        return ExchangeRatesStreamReader(self.options)


class _FetchWindow(InputPartition):
    def __init__(self, fetched_at_ms: int):
        self.fetched_at_ms = fetched_at_ms


class ExchangeRatesStreamReader(DataSourceStreamReader):
    """Fetch one batch of rates per micro-batch from the configured endpoint."""

    def __init__(self, options):
        self.api_url = options.get("api_url", "https://api.example.com/rates")
        self.base_currencies = options.get("base_currencies", "USD").split(",")

    def initialOffset(self) -> dict:
        import time

        return {"fetch_time": int(time.time() * 1000)}

    def latestOffset(self) -> dict:
        import time

        return {"fetch_time": int(time.time() * 1000)}

    def partitions(self, start: dict, end: dict):
        return [_FetchWindow(end.get("fetch_time", 0))]

    def read(self, partition) -> Iterator[Tuple]:
        from datetime import datetime, timezone

        import requests

        fetched_at = datetime.now(timezone.utc)
        for base in self.base_currencies:
            response = requests.get(self.api_url, params={"base": base}, timeout=10)
            rates = response.json().get("rates", {})
            for quote, rate in rates.items():
                yield (base, quote, float(rate), fetched_at)

Two things in this file are load-bearing for the wiring. First, the name() classmethod returns "exchange_rates" — that string, not the class name, becomes the .format(...) argument in the generated read. Second, because readMode will be stream, the class implements streamReader and a DataSourceStreamReader so Spark manages the offsets. A batch source would implement reader / DataSourceReader instead.

Declare the custom data source load¶

A flowgroup is a short YAML file describing a sequence of actions. This one has two — a load that reads through your DataSource, and a write that appends the result into a streaming table. Create pipelines/exchange_rates.yaml:

pipelines/exchange_rates.yaml¶

# Ingest live currency exchange rates from a REST API through a user-supplied
# PySpark DataSource. Action chain: load (custom_datasource) -> write.
pipeline: bronze_rates
flowgroup: exchange_rates

actions:
  # 1. Load: read through the ExchangeRatesDataSource class. LHP copies the
  #    module, registers the class, and reads it as a streaming source.
  - name: load_exchange_rates
    type: load
    readMode: stream
    source:
      type: custom_datasource
      module_path: "data_sources/exchange_rates_source.py"
      custom_datasource_class: ExchangeRatesDataSource
      options:
        api_url: "${exchange_api_url}"
        base_currencies: "USD,EUR,GBP"
    target: v_exchange_rates

  # 2. Write: append the fetched rates into a bronze streaming table.
  - name: write_rates_bronze
    type: write
    source: v_exchange_rates
    write_target:
      type: streaming_table
      catalog: "${catalog}"
      schema: "${bronze_schema}"
      table: exchange_rates

The source block names the class by two required fields: module_path points at the .py file relative to the project root, and custom_datasource_class names the class inside it. Everything under options is handed to your reader — here an api_url and a base_currencies list, which surface as self.options in the DataSourceStreamReader. readMode: stream selects the streaming reader; drop it and stream is the default anyway.

The ${...} tokens are placeholders you resolve per environment, so the same flowgroup runs unchanged in dev, staging, and prod.

Point the tokens at your environment¶

The ${exchange_api_url} in options and the ${catalog} / ${bronze_schema} in the write resolve from a per-environment substitutions file. Create substitutions/dev.yaml:

substitutions/dev.yaml¶

# Development environment tokens.
# Referenced from flowgroups with ${token} syntax.
dev:
  catalog: dev_catalog
  bronze_schema: bronze
  exchange_api_url: https://api.exchangerates.example.com/v1/latest

This is the only place environment-specific values live. Substitution reaches into options and into the copied DataSource file itself, so a production endpoint — or a ${secret:scope/key} reference for an API key — is a change to prod.yaml, never to the flowgroup or the class.

Generate the pipeline¶

Now compile the YAML into Lakeflow code. Validate first, then generate:

$ lhp validate --env dev
✓ discover (0.01s)
✓ preflight (0.00s)
✓ bronze_rates  0 files
✓ validate (0.33s)
1 validated · 0.3s

$ lhp generate --env dev
✓ discover (0.01s)
✓ preflight (0.00s)
✓ bronze_rates  1 file
✓ generate (0.35s)
✓ format (0.05s)
✓ monitoring (0.00s)
1 pipeline generated · 1 file · 0.4s

validate resolves every token and checks the actions before you commit to generating; generate writes the Python — and copies your DataSource module next to it.

Read what Lakehouse Plumber wrote¶

Open generated/dev/bronze_rates/exchange_rates.py. This is the entire output — nothing is hidden behind a runtime:

generated/dev/bronze_rates/exchange_rates.py¶

# Generated by LakehousePlumber
# Pipeline: bronze_rates
# FlowGroup: exchange_rates

from pyspark import cloudpickle as _lhp_cloudpickle
from pyspark import pipelines as dp
from custom_python_functions.exchange_rates_source import ExchangeRatesDataSource
import custom_python_functions

_lhp_cloudpickle.register_pickle_by_value(custom_python_functions)

# Pipeline Configuration
PIPELINE_ID = "bronze_rates"
FLOWGROUP_ID = "exchange_rates"


# ============================================================================
# SOURCE VIEWS
# ============================================================================

# Try to register the custom data source
try:
    spark.dataSource.register(ExchangeRatesDataSource)
except Exception:
    pass  # Ignore if already registered


@dp.temporary_view()
def v_exchange_rates():
    """Load data from custom data source: ExchangeRatesDataSource"""
    df = (
        spark.readStream.format("exchange_rates")
        .option("api_url", "https://api.exchangerates.example.com/v1/latest")
        .option("base_currencies", "USD,EUR,GBP")
        .load()
    )

    return df


# ============================================================================
# TARGET TABLES
# ============================================================================

# Create the streaming table
dp.create_streaming_table(
    name="dev_catalog.bronze.exchange_rates", comment="Streaming table: exchange_rates"
)


# Define append flow(s)
@dp.append_flow(
    target="dev_catalog.bronze.exchange_rates",
    name="f_rates_bronze",
    comment="Append flow to dev_catalog.bronze.exchange_rates",
)
def f_rates_bronze():
    """Append flow to dev_catalog.bronze.exchange_rates"""
    # Streaming flow
    df = spark.readStream.table("v_exchange_rates")

    return df

Every piece of the wiring is here. LHP registered the class with spark.dataSource.register(ExchangeRatesDataSource), read it with spark.readStream.format("exchange_rates") — that format string came from your name() method, not the class name — and passed the resolved api_url and base_currencies through as .option(...) calls. The @dp.temporary_view view feeds a create_streaming_table and an @dp.append_flow that land the rates in dev_catalog.bronze.exchange_rates.

The one line you would never have remembered is _lhp_cloudpickle.register_pickle_by_value(custom_python_functions). LHP copied your module into a custom_python_functions/ package beside the generated file (with a # LHP-SOURCE / DO-NOT-EDIT header) and registered that package with PySpark’s vendored cloudpickle, so the DataSource class survives serialization when the pipeline ships it to the executors. Miss that line by hand and the pipeline fails at run time with a pickling error.

What you just did¶

Twenty-two lines of flowgroup YAML compiled to sixty-two lines of Lakeflow Python. Zero lines of the wiring came from you — not the spark.dataSource.register call, not the format(...).load() read, not the @dp.temporary_view, not the create_streaming_table / append_flow, and not the cloudpickle registration.

The DataSource class stays yours: LHP does not write your fetch logic, and it does not touch the class it copies. What it owns is the plumbing that turns a class on disk into a registered, executor-safe, wired-up streaming source — the part that is fiddly to get right and identical every time.

What’s next¶

Switch to a batch source. Set readMode: batch and implement reader / DataSourceReader on the class instead of the streaming pair; LHP emits spark.read.format(...) in place of spark.readStream.
Feed a secret in. Reference ${secret:scope/key} inside options for an API key or token — it resolves to a dbutils.secrets.get(...) call in the generated read, so no credential lands in YAML.
See every field. module_path, custom_datasource_class, options, and the readMode rules are all covered in the load action reference.