Build a custom connector for Lakeflow Connect

Beta

This feature is in Beta. Workspace admins can control access to this feature from the Previews page. See Manage Databricks previews.

Custom connectors let you ingest data from a source that Lakeflow Connect doesn't support with a managed connector. You build and test your connector, then deploy and run it in your own Databricks workspace. You don't need to register it with the community or contribute it to any shared repository to use it.

Develop your connector using the tools and templates in the Lakeflow Community Connectors repository on GitHub. The repository includes AI-powered development tools to assist with each phase, including source research, authentication setup, implementation, and testing.

If you later want to share your connector with other users, you can contribute it to the community. To use an existing community connector, see Community connectors in Lakeflow Connect.

Requirements

Before you start, make sure you have:

• Python 3.10 or above
• A Databricks workspace with Unity Catalog enabled
• API credentials for the source you want to connect to
• Git installed locally

Set up the repository

Clone the Lakeflow Community Connectors repository and install the development dependencies.

1. Clone the repository:

   Bash

   git clone https://github.com/databrickslabs/lakeflow-community-connectors.git
   cd lakeflow-community-connectors

2. Create a virtual environment and install dependencies:

   Bash

   python -m venv .venv
   source .venv/bin/activate
   pip install -e ".[dev]"

3. Review the existing connector implementations in src/databricks/labs/community_connector/sources/, then start developing your connector in a new directory under that path. Follow the repository's AI-assisted development commands and skills. For the recommended workflow, use:

   Text

   /develop-connector <your-source>
   /validate-connector <your-source>

Implement the LakeflowConnect interface

Each connector implements the LakeflowConnect interface, which defines how your connector authenticates, discovers tables, returns schemas, and reads data.

Python

class LakeflowConnect:
    def __init__(self, options: dict[str, str]) -> None:
        """Initialize with connection parameters"""

    def list_tables(self) -> list[str]:
        """Return names of all tables supported by this connector."""

    def get_table_schema(self, table_name: str, table_options: dict[str, str]) -> StructType:
        """Return the Spark schema for a table."""

    def read_table_metadata(self, table_name: str, table_options: dict[str, str]) -> dict:
        """Return metadata: primary_keys, cursor_field, ingestion_type
        (snapshot|cdc|cdc_with_deletes|append)."""

    def read_table(self, table_name: str, start_offset: dict,
                   table_options: dict[str, str]) -> (Iterator[dict], dict):
        """Yield records as JSON dicts and return the next offset
        for incremental reads."""

    def read_table_deletes(self, table_name: str, start_offset: dict,
                           table_options: dict[str, str]) -> (Iterator[dict], dict):
        """Optional: Only required if ingestion_type is 'cdc_with_deletes'."""

Method descriptions

The following table describes each method in the LakeflowConnect interface:

Method	Description
__init__	Receives the connection parameters as a dictionary and initializes the API client for your source.
list_tables	Returns the names of all tables (or API endpoints) your connector exposes. Databricks uses this list to populate the table selection UI.
get_table_schema	Returns a Spark StructType describing the schema for the given table. Called before the first pipeline run and on each run when schema evolution is enabled.
read_table_metadata	Returns a dictionary with primary_keys, cursor_field, and ingestion_type. The ingestion_type must be one of snapshot, cdc, cdc_with_deletes, or append.
read_table	Yields records as Python dictionaries and returns the next offset for incremental reads. On the first run, start_offset is empty. On subsequent runs, it contains the offset returned by the previous run.
read_table_deletes	Optional. Only implement this method if ingestion_type is cdc_with_deletes. Yields deleted record keys and returns the next offset.

Method	Description
__init__	Receives the connection parameters as a dictionary and initializes the API client for your source.
list_tables	Returns the names of all tables (or API endpoints) your connector exposes. Databricks uses this list to populate the table selection UI.
get_table_schema	Returns a Spark StructType describing the schema for the given table. Called before the first pipeline run and on each run when schema evolution is enabled.
read_table_metadata	Returns a dictionary with primary_keys, cursor_field, and ingestion_type. The ingestion_type must be one of snapshot, cdc, cdc_with_deletes, or append.
read_table	Yields records as Python dictionaries and returns the next offset for incremental reads. On the first run, start_offset is empty. On subsequent runs, it contains the offset returned by the previous run.
read_table_deletes	Optional. Only implement this method if ingestion_type is cdc_with_deletes. Yields deleted record keys and returns the next offset.

Develop your connector

Follow these steps to build and validate a new connector:

1. Research the source API: Study the source's API specifications, authentication mechanisms, rate limits, and available data schemas. Identify which tables or endpoints to expose.

2. Set up authentication: Generate the connection specification, configure credentials for the source, and verify connectivity from your development environment.

3. Implement the connector: Code all required LakeflowConnect interface methods to connect to the source API and return data in the expected format.

4. Test and iterate: Run the standard test suites against a real source system and fix any issues. See Test your connector for details.

5. Document the connector: Write a user-facing README.md and generate the connector spec YAML file that describes the connector's configurable parameters.

6. Build the deployment artifact: Run the build script to produce the single-file artifact that can be deployed in a workspace.

Test your connector

The repository provides several testing approaches:

Generic test suite (required)

This suite connects to a real source using your provided credentials to verify end-to-end functionality, including authentication, schema discovery, and data reads.

Bash

python -m pytest tests/generic/ --connector <your-source> --credentials credentials.json

Write-back testing (recommended)

Write-back testing runs write-read-verify cycles to validate incremental reads and deletes. This confirms that your offset tracking and CDC logic work correctly.

Bash

python -m pytest tests/writeback/ --connector <your-source> --credentials credentials.json

Unit tests

Write unit tests for any complex custom logic in your connector, such as pagination handling, type coercion, or error recovery.

Build the deployment artifact

After your connector passes the test suites, package it so a pipeline can run it. A connector deploys as two parts:

• A single-file source artifact. Run the merge script to flatten your connector into one self-contained Python file. The pipeline uses this file at runtime rather than the full repository.

  Bash

  python tools/scripts/merge_python_source.py --connector <your-source>

  The script writes this file to dist/<your-source>/.

• Python wheels (.whl) for the connector's dependencies. The connector framework and any third-party libraries your connector imports must be available to the pipeline as wheels stored in a Unity Catalog volume. If these wheels are missing, the pipeline can fail during source discovery. You can upload them yourself and reference them in the UI, or let the Community Connector CLI build and upload them for you. See Deploy with the Community Connector CLI.

Deploy your connector in one of two ways:

• The Databricks UI is the point-and-click path. Use it for a one-off deployment when you provide the connector's wheels yourself in the Library dependencies field. See Deploy in the Databricks UI.
• The community-connector CLI is the scriptable path. Use it when you're developing locally and want the connector's wheels built and uploaded for you, or when you want a repeatable deployment you can automate. See Deploy with the Community Connector CLI.

Deploy in the Databricks UI

Deploy your connector in the Databricks UI in two phases: add the connector, then create the pipeline.

Add the custom connector

First, add your connector so that it appears as a tile on the Add data page:

1. In the sidebar of your Databricks workspace, click +New > Add or upload data, then, under Community connectors, add a custom connector.
2. For Source name, enter the name of your connector. This must match the directory name that contains your connector's source code (sources/<source-name>).
3. For Display name, enter a friendly name for the connector. If you leave this blank, it defaults to the source name.
4. For Library dependencies, add the Python wheel (.whl) files your connector needs from a Unity Catalog volume. See Build the deployment artifact.
5. For Connection specification, paste the connector's connection specification in YAML, matching its connector_spec.yaml file.
6. Click Save. The connector appears as a Custom tile under Community connectors.

Create the ingestion pipeline

Then create the pipeline that ingests data from your source:

1. Select your connector tile to open the Ingest data wizard.
2. On the Connection step, click + Create connection or select an existing connection, enter the connection details for your source, then click Next.
3. On the Ingestion setup step, enter a Pipeline name, set the Event log location (catalog and schema), choose a Compute type, then click Create pipeline and continue.
4. On the Source step, select the tables to ingest.
5. On the Destination step, choose the catalog and schema where ingested tables are written.
6. On the Schedules and notifications step, set an optional schedule and notifications, then finish.
7. Run the pipeline manually or on its schedule.

To configure the pipeline further, you can edit ingest.py in the pipeline editor. See Pipeline configuration options.

Pipeline configuration options

You can configure the following options in ingest.py:

Option	Description
connection_name	Required. The name of the connection that stores authentication credentials for the source.
objects	Required. A list of tables to ingest. Each entry has the format {"table": {"source_table": "..."}}. You can also specify an optional destination_table inside the table object.
destination_catalog	The catalog where ingested tables are written. Defaults to the catalog set during pipeline creation.
destination_schema	The schema where ingested tables are written. Defaults to the schema set during pipeline creation.
scd_type	The slowly changing dimension strategy: SCD_TYPE_1, SCD_TYPE_2, or APPEND_ONLY. Defaults to SCD_TYPE_1.
primary_keys	Override the default primary keys for a table. Provide a list of column names.

Option	Description
connection_name	Required. The name of the connection that stores authentication credentials for the source.
objects	Required. A list of tables to ingest. Each entry has the format {"table": {"source_table": "..."}}. You can also specify an optional destination_table inside the table object.
destination_catalog	The catalog where ingested tables are written. Defaults to the catalog set during pipeline creation.
destination_schema	The schema where ingested tables are written. Defaults to the schema set during pipeline creation.
scd_type	The slowly changing dimension strategy: SCD_TYPE_1, SCD_TYPE_2, or APPEND_ONLY. Defaults to SCD_TYPE_1.
primary_keys	Override the default primary keys for a table. Provide a list of column names.

Deploy with the Community Connector CLI

The CLI's publish command builds the framework and connector wheels from your local source, uploads them to a Unity Catalog volume, records their paths in the connector manifest, and publishes the connector as a Custom tile on the Add data page. Point it at your local connector spec so it doesn't look for the connector in the upstream repository:

Bash

community-connector publish <your-source> \
  --spec src/databricks/labs/community_connector/sources/<your-source>/connector_spec.yaml

To reuse wheels you already built and skip the build step, pass them with --package. To replace a connector you published earlier, add --overwrite. For the full list of options, including --package, --volume-path, --catalog, and --schema, see the publish command reference.

To run the entire workflow from the command line, including creating a connection, creating and updating the ingestion pipeline, publishing, and unpublishing, see the Community Connector CLI reference.

Contribute your connector to the community

Your connector runs in your workspace whether or not you contribute it. If you want to share it so other users can discover and use it, open a pull request in the Lakeflow Community Connectors repository. Contributed connectors become community connectors, which the community maintains and which aren't backed by Databricks SLAs.