Load Data from Airbyte Sources
Load data from any Airbyte Python CDK source into LlamaIndex, with lazy and incremental loading.
Why it matters
Seamlessly ingest data from any Airbyte source into LlamaIndex for advanced AI applications. This asset acts as a bridge, enabling your AI to access and process information from diverse data streams.
Outcomes
What it gets done
Connect to and extract data from any source compatible with the Airbyte CDK.
Load data into LlamaIndex documents, with flexible text and metadata handling.
Support for incremental data loading to efficiently update AI knowledge bases.
Enable lazy loading for handling large datasets without memory constraints.
Install
Add it to your toolbox
Run in your project directory:
curl -fsSL https://spark.entire.vc/get/li-reader-readers-airbyte-cdk | bash Overview
Airbyte CDK Loader
A LlamaIndex reader that wraps any Airbyte Python CDK source, with lazy loading for large streams and incremental sync via trackable state. Use for any Airbyte-supported source, choosing lazy loading for large streams and incremental sync for recurring runs.
What it does
The Airbyte CDK Loader is a shim for sources built with the Airbyte Python CDK, letting LlamaIndex load data from any Airbyte source. Rather than reimplementing connector logic, it wraps an existing or custom Airbyte CDK source class so its data becomes available as LlamaIndex documents.
AirbyteCDKReader is initialized with a source_class (an Airbyte source, such as SourceGithub imported from its own package) and a config dictionary matching that source's expected configuration, and load_data takes a stream_name identifying which stream to pull from that source. By default, every field from each record is stored as document metadata, and the document's text is the JSON representation of all fields; a custom record_handler function can be passed to the reader to construct the document text and metadata differently - for example extracting just a title field as the text.
Two loading modes are supported beyond the standard load_data. lazy_load_data returns an iterator instead of a full list, so documents can be consumed one at a time without holding everything in memory - useful for large record sets. Incremental loading, when the underlying stream supports it, lets the reader track state (reader.last_state, which can be pickled and stored) and pass that state back into a later load_data call so only new or updated records since the last run are returned, rather than reloading everything.
When to use - and when NOT to
Use it when you need to bring data from any Airbyte-supported source into LlamaIndex without writing a custom connector - install the relevant Airbyte source package, point the reader at its source class and config, and pick a stream. Use lazy_load_data for large streams to avoid memory pressure, and incremental loading (tracking last_state) for recurring syncs where you only want new or changed records each run. Do not use it if the target source has no existing Airbyte CDK implementation; you would need to implement one first using Airbyte's own connector-development documentation.
Capabilities
load_data/lazy_load_data pull records from a named stream of any Airbyte CDK source, with default JSON-as-text/all-fields-as-metadata document construction, an optional custom record_handler, and incremental loading via a trackable last_state.
How to install
pip install llama-index-readers-airbyte-cdk
Also requires installing airbyte-cdk and the specific Airbyte source package for the data source you want to load from.
Who it's for
Developers who need to load data from any Airbyte-supported source into LlamaIndex, especially for large record sets (via lazy loading) or recurring incremental syncs, without writing a custom connector.
Source README
Airbyte CDK Loader
pip install llama-index-readers-airbyte-cdk
The Airbyte CDK Loader is a shim for sources created using the Airbyte Python CDK. It allows you to load data from any Airbyte source into LlamaIndex.
Installation
- Install llama-index reader:
pip install llama-index-readers-airbyte-cdk - Install airbyte-cdk:
pip install airbyte-cdk - Install a source via git (or implement your own):
pip install git+https://github.com/airbytehq/airbyte.git@master#egg=source_github&subdirectory=airbyte-integrations/connectors/source-github
Usage
Implement and import your own source. You can find lots of resources for how to achieve this on the Airbyte documentation page.
Here's an example usage of the AirbyteCdkReader.
from llama_index.readers.airbyte_cdk import AirbyteCDKReader
from source_github.source import (
SourceGithub,
) # this is just an example, you can use any source here - this one is loaded from the Airbyte Github repo via pip install git+https://github.com/airbytehq/airbyte.git@master#egg=source_github&subdirectory=airbyte-integrations/connectors/source-github`
github_config = {
# ...
}
reader = AirbyteCDKReader(source_class=SourceGithub, config=github_config)
documents = reader.load_data(stream_name="issues")
By default all fields are stored as metadata in the documents and the text is set to the JSON representation of all the fields. Construct the text of the document by passing a record_handler to the reader:
def handle_record(record, id):
return Document(
doc_id=id, text=record.data["title"], extra_info=record.data
)
reader = AirbyteCDKReader(
source_class=SourceGithub,
config=github_config,
record_handler=handle_record,
)
Lazy loads
The reader.load_data endpoint will collect all documents and return them as a list. If there are a large number of documents, this can cause issues. By using reader.lazy_load_data instead, an iterator is returned which can be consumed document by document without the need to keep all documents in memory.
Incremental loads
If a stream supports it, this loader can be used to load data incrementally (only returning documents that weren't loaded last time or got updated in the meantime):
reader = AirbyteCDKReader(source_class=SourceGithub, config=github_config)
documents = reader.load_data(stream_name="issues")
current_state = reader.last_state # can be pickled away or stored otherwise
updated_documents = reader.load_data(
stream_name="issues", state=current_state
) # only loads documents that were updated since last time
This loader is designed to be used as a way to load data into LlamaIndex.
FAQ
Common questions
Discussion
Questions & comments · 0
Sign In Sign in to leave a comment.