Ingest Designer Guide

• 1: Writing a Config
• 2: Defining Pipelines
• 3: Control Flow
• 4: Connectors
• 4.1: File Connector
• 4.2: Database Connector
• 4.3: Solr Connector
• 4.4: Kafka Connector
• 4.5: Parquet Connector
• 4.6: RSS Connector
• 5: Stages
• 5.1: All Stages
• 5.2: ChunkText
• 5.3: PromptOllama
• 5.4: QueryOpensearch
• 5.5: EmbeddedPython
• 5.6: ExternalPython
• 6: Indexers
• 6.1: Solr Indexer
• 6.2: OpenSearch Indexer
• 6.3: Elasticsearch Indexer
• 6.4: CSV Indexer
• 6.5: NopIndexer (No-op)
• 6.6: Pinecone Indexer
• 6.7: Weaviate Indexer
• 7: Cookbooks
• 7.1: File Ingestion Cookbook
• 7.2: Vector Search Cookbook
• 7.3: RSS Cookbook

1 - Writing a Config

When you run Lucille, you provide a path to a configuration file that defines your entire ingest: which sources to read from, which pipelines to apply, and where to send the results. Configuration files use HOCON, a superset of JSON.

Quick references:

• s3-opensearch.conf — A simple, runnable example that ingests files from S3 into OpenSearch.
• application-example.conf — A comprehensive, annotated illustration of every top-level property and block that a Lucille config can contain (worker, publisher, kafka, indexer, etc.), excluding the implementation-specific parameters of individual stages, connectors, and indexers. Use this as a reference when you need to know what options exist and what they do. Note: this file is a reference document, not a tested runnable config.
• lightbend/config — The upstream documentation for HOCON syntax and the Typesafe Config library that Lucille uses to parse configs. Consult this when you need the definitive rules on substitution syntax, include semantics, value concatenation, or other HOCON language features beyond what this guide covers.

Required Elements

A complete config file must contain three elements:

Connectors

Connectors read data from a source and emit it as a sequence of individual Documents, which are then sent to a Pipeline for enrichment.

connectors should be populated with a list of Connector configurations.

connectors: [
  {
    name: "my-connector"
    class: "com.kmwllc.lucille.connector.FileConnector"
    pipeline: "my-pipeline"
    paths: ["/data/files"]
  }
]

See Connectors for the full catalogue of available connectors and their parameters.

Pipelines and Stages

A pipeline is a list of Stages applied to incoming Documents, preparing them for indexing. Each Connector specifies which pipeline will process its output via the pipeline field.

pipelines should be populated with a list of Pipeline configurations. Each Pipeline needs a name and a stages list. Multiple connectors may feed the same Pipeline.

pipelines: [
  {
    name: "my-pipeline"
    stages: [
      { class: "com.kmwllc.lucille.stage.RenameFields", fieldMapping: { old_name: new_name } }
      { class: "com.kmwllc.lucille.stage.TrimWhitespace", fields: ["title", "description"] }
    ]
  }
]

See Stages for the full catalogue of available stages and their parameters.

Indexer

An Indexer sends processed Documents to a destination — typically a search engine. Only one Indexer can be defined per run; all pipelines feed to the same Indexer.

A full indexer configuration has two separate config blocks: the generic indexer block and a backend-specific block (e.g., solr, opensearch, elastic).

indexer {
  type: "OpenSearch"
  batchSize: 100
  batchTimeout: 500
}

opensearch {
  url: "https://localhost:9200"
  index: "my-index"
}

See Indexers for available indexer types and their configuration.

Other Run Configuration

In addition to the three required elements, you can configure other parts of a Lucille run. See application-example.conf for the complete annotated reference of all available top-level options.

Validation

Lucille validates the configuration for every Connector, Stage, and Indexer before starting a run. If you provide an unrecognized property, or omit a required one, Lucille throws an exception at startup rather than discovering the problem mid-run.

To validate your config without starting a run, use the -validate flag:

java -Dconfig.file=my-config.conf -cp '...' com.kmwllc.lucille.core.Runner -validate

All errors are reported together — not just the first one — so you can fix them all at once.

Each built-in component declares a SPEC that enumerates its legal configuration properties. The validator checks your config against these SPECs. For details on how validation works internally and how to declare SPECs for custom components, see SPEC Validation System.

To see the fully resolved config (with all substitutions applied), use the -render flag:

java -Dconfig.file=my-config.conf -cp '...' com.kmwllc.lucille.core.Runner -render

HOCON Basics for Ingest Designers

HOCON is a superset of JSON with features that make config files more readable and maintainable:

• Comments — use # or // to annotate your config
• Relaxed syntax — quotes around keys are optional, trailing commas are allowed
• Environment variable substitution — inject credentials and environment-specific values without hardcoding them:

opensearch {
  url: "http://localhost:9200"
  url: ${?OPENSEARCH_URL}   # overrides with env var if set
}

• Includes — compose configs from reusable fragments:

include "shared-stages.conf"

• Internal references — reuse values defined elsewhere in the same file:

common { batchSize: 1000 }
indexer { batchSize: ${common.batchSize} }

For a deeper treatment of HOCON patterns, environment variable substitution, and config composition, see Configuration Management in the Operations Guide.

2 - Defining Pipelines

Defining a Pipeline

Pipelines are defined in the pipelines list in your config file. Each pipeline requires a name and a stages list:

pipelines: [
  {
    name: "my-pipeline"
    stages: [
      {
        class: "com.kmwllc.lucille.stage.RenameFields"
        fieldMapping: { old_name: new_name }
      },
      {
        class: "com.kmwllc.lucille.stage.TrimWhitespace"
        fields: ["title", "description"]
      }
    ]
  }
]

Stages execute in the order they are listed. Each Stage receives the Document as mutated by all previous Stages.

Connecting a Connector to a Pipeline

Each Connector specifies which pipeline will process its output via the pipeline field:

connectors: [
  {
    name: "my-connector"
    class: "com.kmwllc.lucille.connector.FileConnector"
    pipeline: "my-pipeline"
    paths: ["/data/files"]
  }
]

Multiple Connectors can feed the same Pipeline. When two connectors reference the same pipeline name, they share the pipeline definition but not the pipeline instance. Each connector’s execution creates a fresh WorkerPool with new Stage instances, runs to completion, and tears down before the next connector begins. State accumulated in Stage instance fields (database connections, counters, caches) does not carry over between connectors — each connector starts with a clean pipeline.

The pipeline field is optional. A connector without a pipeline performs its work without publishing documents — useful for preparatory tasks like creating an index or running a migration before a publishing connector executes. See Setup-Only Connectors for details.

Multiple Pipelines in a Single Run

You can define multiple pipelines in a single config, each serving different connectors:

connectors: [
  { name: "csv-connector",  class: "...", pipeline: "csv-pipeline" },
  { name: "json-connector", class: "...", pipeline: "json-pipeline" }
]

pipelines: [
  { name: "csv-pipeline",  stages: [...] },
  { name: "json-pipeline", stages: [...] }
]

All Pipelines feed the same Indexer.

Empty Pipeline

A Pipeline with an empty stages list is valid. Documents pass through without transformation and are sent directly to the Indexer:

pipelines: [
  { name: "passthrough", stages: [] }
]

Conditional Stage Execution

Every Stage supports a conditions block that controls whether the Stage applies to a given Document. Conditions can check field presence, field values, or combinations:

{
  class: "com.kmwllc.lucille.stage.OpenAIEmbed"
  fields: ["content"]
  conditions: [
    { fields: ["content_type"], values: ["article"] }
  ]
}

In this example, OpenAIEmbed only runs on Documents where content_type equals "article".

Use conditionPolicy to control how multiple conditions combine:

• "all" (default) — all conditions must be true
• "any" — at least one condition must be true

See Stages for the full conditions reference.

Reusing Stage Sequences Across Pipelines

If you want to share a common sequence of stages across multiple pipelines, define them in a separate config file and compose them using HOCON’s array concatenation:

include "shared-stages.conf"   # defines: shared_stages = [{...}, {...}]

pipelines: [{
  name: "my-pipeline"
  stages = [{ name: "first_stage", class: "..." }] ${shared_stages} [{ name: "last_stage", class: "..." }]
}]

HOCON concatenates adjacent arrays into a single list, so the resolved stages array contains first_stage, followed by the shared stages, followed by last_stage.

Testing Your Ingest

You can verify that your pipeline produces the expected output by running it in test mode. Lucille’s test mode executes the full pipeline end-to-end — real connectors, real stages, real document routing — but bypasses the search backend and captures all output in memory for assertion. This requires writing a short Java test. See Testing Pipelines for a complete guide.

For config-only validation (no Java required), use the -validate flag described in Writing a Config.

Building a Pipeline From Code

For integration tests or programmatic usage, pipelines can be constructed from a Config object:

Config config = ConfigFactory.load("my-config.conf");
Pipeline pipeline = Pipeline.fromConfig(config, "my-pipeline");
pipeline.startStages();

Iterator<Document> results = pipeline.processDocument(doc);

pipeline.stopStages();

See Testing Pipelines for more on testing patterns.

3 - Control Flow

This page is relevant to both ingest designers (working in config only) and component developers (writing Java stage and connector code). Each section identifies which approach applies to you.

Conditional Stage Execution

Config | When to use: You want a stage to run only on documents that have a particular field, or a particular value in a field. For example, only run an embedding stage on documents where content_type is "article", or only run a cleanup stage on documents that have a raw_html field.

Add a conditions block to the stage in your config. The base Stage class evaluates conditions before calling processDocument() — if the conditions are not met, the stage is skipped for that document entirely. You never need to implement this logic inside a stage.

{
  class: "com.kmwllc.lucille.stage.OpenAIEmbed"
  fields: ["content"]
  dest: ["content_vector"]
  apiKey: ${OPENAI_API_KEY}
  conditionPolicy: "all"
  conditions: [
    { fields: ["content"] }
    { fields: ["content_type"], values: ["article"] }
  ]
}

See Stages for the full conditions reference including operator and conditionPolicy.

Skipping a Document

Config + Code | When to use: You want a document to bypass all remaining pipeline stages but still be sent to the indexer. The canonical use case is a deletion tombstone: a connector emits a document representing a record that should be deleted from the index, and you want it to reach the indexer without being enriched or transformed by downstream stages.

Config approach: Add a SkipDocument stage to your pipeline with conditions that identify which documents should be skipped.

{
  class: "com.kmwllc.lucille.stage.SkipDocument"
  conditions: [
    { fields: ["is_tombstone"], values: ["true"] }
  ]
}

Code approach: From inside processDocument(), call doc.setSkipped(true). All subsequent stages in the pipeline check isSkipped() before processing and will skip the document automatically.

In both cases the document still flows to the indexer.

Dropping a Document

Config + Code | When to use: You want to abandon a document entirely — it should not be processed by downstream stages and should not be sent to the indexer. For example, you might drop documents that fail a quality check, documents that represent records you don’t need to reprocess, or documents that are out of scope for the current run.

Config approach: Add a DropDocument stage to your pipeline with conditions that identify which documents should be dropped.

{
  class: "com.kmwllc.lucille.stage.DropDocument"
  conditions: [
    { fields: ["status"], values: ["archived"] }
  ]
}

Code approach: From inside processDocument(), call doc.setDropped(true). The document will not be processed by any subsequent stage and will not be sent to the indexer. Lucille records it as a dropped document in the run summary.

DropDocument also accepts an optional percentage parameter (a value between 0.0 and 1.0) that drops documents probabilistically. This is primarily useful for testing to simulate document loss in a non-deterministic way.

Sending a Document to an Error State

Code | When to use: You are writing a stage and something has gone wrong that makes it impossible or unsafe to continue processing the document — for example, a required external service is unavailable, or a field contains data in a format the stage cannot handle. You want the document to stop processing, not be indexed, and be recorded as a failure in the run summary.

Throw a StageException from processDocument().

@Override
public Iterator<Document> processDocument(Document doc) throws StageException {
  String value = doc.getString("required_field");
  if (value == null) {
    throw new StageException("required_field is missing on document " + doc.getId());
  }
  // ...
}

The framework catches the exception, marks the document as failed, and continues processing other documents. The run summary will report the document as an error.

Use this conservatively. Most stages should not throw StageException for ordinary data variation — a missing optional field or an unexpected value is usually better handled by logging or by skipping the transformation for that document. Reserve StageException for conditions where continuing to process the document would produce incorrect or corrupt results.

Logging an Error Without Stopping Processing

Code | When to use: Something unexpected happened while processing a document, but it is not severe enough to stop processing or prevent indexing. You want to record the problem for later investigation — in the logs, in the index itself, or both.

Log the error using the stage’s logger, and optionally write a custom field to the document that will be visible in the index.

@Override
public Iterator<Document> processDocument(Document doc) throws StageException {
  try {
    String result = callExternalService(doc.getString("content"));
    doc.setField("enriched_content", result);
  } catch (Exception e) {
    log.error("External service call failed for document {}: {}", doc.getId(), e.getMessage());
    doc.setField("enrichment_error", e.getMessage());
    // processing continues; document will still be indexed
  }
  return null;
}

Adding a dedicated error field to the document (like enrichment_error above) makes failures searchable and auditable in the index, which is useful when you need to identify and reprocess documents that encountered a specific problem.

Attaching a Child Document

Code | When to use: You want to associate additional structured data with a document — for example, metadata extracted from a file, or sub-records parsed from a parent record — but you do not want those sub-records to be processed independently by downstream stages or indexed as separate documents. The child data travels with the parent and is accessible to stages that specifically look for it.

Call doc.addChild(childDoc) from inside processDocument() and return null.

@Override
public Iterator<Document> processDocument(Document doc) throws StageException {
  Document child = Document.create(doc.getId() + "-metadata");
  child.setField("extracted_author", parseAuthor(doc.getString("raw_header")));
  doc.addChild(child);
  return null;
}

Attached children are stored inside the parent under the reserved ___children field. Ordinary stages ignore them. Only stages that explicitly iterate doc.getChildren() will see them. They are not independently tracked by the Publisher and are not indexed as separate records unless a subsequent EmitNestedChildren stage promotes them.

Emitting a Child Document

Config + Code | When to use: You want to produce additional documents that flow through the remaining pipeline stages independently and are indexed as separate records. The canonical use case is chunking: a single large document is split into many smaller chunks, each of which needs to be embedded and indexed on its own. More generally, any time a stage produces attached children (via doc.addChild()) and you want those children to become independent pipeline documents, you need to emit them.

Config approach: Add an EmitNestedChildren stage to your pipeline after any stage that attaches children. It detaches the children from the parent and emits them as independent documents. If the parent document itself should not be indexed, set dropParent: true. Use fieldsToCopy to copy fields from the parent to each child before they separate — useful for propagating metadata like a title or source URL.

{
  class: "com.kmwllc.lucille.stage.EmitNestedChildren"
  dropParent: true
  fieldsToCopy: {
    "title": "parent_title"
    "source_url": "source_url"
  }
}

EmitNestedChildren is a no-op if the document has no attached children, so it is safe to place in a pipeline where only some documents will have children. See ChunkText for the common chunking pattern that uses this stage.

Code approach: Return an Iterator<Document> directly from processDocument(). Children returned this way become independent pipeline documents immediately — they are tracked by the Publisher, processed by all downstream stages, and indexed separately. No EmitNestedChildren stage is needed.

@Override
public Iterator<Document> processDocument(Document doc) throws StageException {
  List<String> chunks = chunk(doc.getString("body"));
  List<Document> children = new ArrayList<>();
  for (int i = 0; i < chunks.size(); i++) {
    Document child = Document.create(doc.getId() + "-chunk-" + i);
    child.setField("text", chunks.get(i));
    child.setField("parent_id", doc.getId());
    children.add(child);
  }
  return children.iterator();
}

Running Connectors in Sequence

Config | When to use: You have two or more connectors that must run in a specific order, and the second should only run if the first succeeds. For example, a connector that deletes stale records from the index followed by a connector that re-ingests fresh records — you don’t want the re-ingest to proceed if the deletion step failed.

List the connectors in order in the connectors array of a single Lucille config. Lucille runs connectors sequentially and aborts the run if any connector fails (meaning any of its lifecycle methods throw an exception). A connector is not considered failed if individual documents it publishes encounter errors during pipeline processing.

connectors: [
  {
    name: "delete-stale"
    class: "com.kmwllc.lucille.connector.SolrConnector"
    // ...
  },
  {
    name: "ingest-fresh"
    class: "com.kmwllc.lucille.connector.FileConnector"
    pipeline: "my-pipeline"
    // ...
  }
]

Note that all connectors in a single config share the same indexer. If your connectors need to write to different backends, use separate configs.

Running Connectors in Parallel

Config | When to use: You have two independent ingests that do not depend on each other and you want them to run at the same time to reduce total wall-clock time.

Create two separate Lucille config files and launch them as separate processes from the command line. Each process runs its own Runner, its own pipeline workers, and its own indexer connection independently.

java -Dconfig.file=ingest-a.conf -cp '...' com.kmwllc.lucille.core.Runner &
java -Dconfig.file=ingest-b.conf -cp '...' com.kmwllc.lucille.core.Runner &

There is no built-in mechanism for parallel connector execution within a single Lucille process. If the two ingests write to the same index, ensure their document IDs do not collide.

Pre- and Post-Connector Actions

Config + Code | When to use: You need to perform setup or teardown work that is tightly coupled to a connector’s execution — for example, deleting stale records from a search index before a connector re-ingests them, or committing the index after ingestion completes. You want this work to be part of the connector’s lifecycle so that failures in setup prevent the connector from running at all.

Config approach: SolrConnector is the canonical example. It accepts preActions and postActions lists in its config — arbitrary Solr update requests (JSON or XML) that are issued before and after the connector queries Solr. The {runId} placeholder in action strings is substituted with the current run ID at execution time.

{
  name: "solr-reindex"
  class: "com.kmwllc.lucille.connector.SolrConnector"
  pipeline: "my-pipeline"
  solr: { url: "http://localhost:8983/solr/mycore" }
  preActions: [
    "{\"delete\": {\"query\": \"runId:{runId}\"}}"
  ]
  postActions: [
    "{\"commit\": {}}"
  ]
}

Code approach: Override preExecute(String runId) and/or postExecute(String runId) in your connector implementation. The lifecycle contract is:

• preExecute() is always called first. If it throws a ConnectorException, execute() and postExecute() are skipped.
• execute() is called if preExecute() succeeds. If it throws, postExecute() is skipped.
• postExecute() is called only if both preExecute() and execute() succeed.
• close() is always called regardless of what happened above.

Keep resource cleanup (closing connections, releasing file handles) in close(), not postExecute(). postExecute() is for post-success actions; close() is the guaranteed teardown.

Setup-Only Connectors (No Pipeline)

Config + Code | When to use: You need to perform preparatory work — creating a collection, running a database migration, clearing an index, calling an external API — before a publishing connector runs. You want this work to be part of the run so that failures prevent downstream connectors from executing.

How it works: Omit the pipeline field from the connector config. When a connector has no pipeline, the framework skips creating a Publisher, WorkerPool, and Indexer. It calls execute() synchronously with a null publisher. The connector’s execute() method performs its work and returns. If it throws, the run is aborted and subsequent connectors do not execute.

connectors: [
  {
    name: "prepare-index"
    class: "com.example.CreateCollectionConnector"
    # no pipeline field — this connector does not publish documents
    solrUrl: "http://localhost:8983/solr"
    collection: "my-collection"
  },
  {
    name: "ingest-data"
    class: "com.kmwllc.lucille.connector.FileConnector"
    pipeline: "enrichment-pipeline"
    paths: ["/data/source"]
  }
]

In this example, prepare-index runs first. If it succeeds, ingest-data runs and publishes documents into the pipeline. If prepare-index throws, the run aborts and ingest-data never starts.

Lifecycle behavior for a no-pipeline connector:

• preExecute(runId) — called normally
• execute(null) — called synchronously; the publisher argument is null
• postExecute(runId) — called if execute succeeds
• close() — always called

When to use this vs. preActions: Use a setup-only connector when the preparatory logic is complex, reusable across runs, or not tied to a specific connector’s source system. Use preActions/postActions (on connectors that support them, like SolrConnector) when the setup is a simple command tightly coupled to that connector’s execution.

Stage Initialization and Teardown

Code | When to use: You are writing a stage that needs to acquire resources before processing begins — opening a database connection, building an HTTP client, loading a model file, compiling a regex or expression, or validating configuration values that can only be checked at runtime. Override start() and/or stop() to manage that lifecycle cleanly.

start()

start() is called once per worker thread, after the stage is constructed but before any documents are processed. Because each worker thread gets its own pipeline instance with its own stage instances, resources allocated in start() are effectively thread-local — no synchronization is needed.

Use start() for:

• Opening connections — database connections, HTTP clients, external service clients. QueryDatabase opens its JDBC connection and prepares its SQL statement here; FetchUri builds its CloseableHttpClient here.
• Loading or compiling resources — parsing expressions, loading NLP models, reading lookup files. ApplyJSONata compiles its JSONata expression here; ChunkText loads its OpenNLP sentence model here.
• Runtime config validation — checks that can’t be done in the constructor because they depend on the interaction of multiple config values, or because they involve I/O. FetchUri validates its statusCodeRetryList here and throws StageException if the configuration would cause infinite retries. RenameFields throws if fieldMapping is empty.

Throwing StageException from start() prevents the pipeline from starting at all for that worker thread, which is the right behavior when the stage cannot function without the resource it failed to acquire.

@Override
public void start() throws StageException {
  try {
    this.connection = DriverManager.getConnection(connectionString, jdbcUser, jdbcPassword);
    this.preparedStatement = connection.prepareStatement(sql);
  } catch (SQLException e) {
    throw new StageException("Could not connect to database", e);
  }
}

stop()

stop() is called once per worker thread after all documents have been processed, during pipeline shutdown. Use it to release whatever start() acquired.

Use stop() for:

• Closing connections — JDBC connections, HTTP clients, script engine contexts. QueryDatabase closes its connection and prepared statement here; FetchUri closes its HTTP client here; ApplyJavascript closes its GraalVM context here.
• Flushing output — Print closes its writer here if one was opened.

@Override
public void stop() throws StageException {
  try {
    if (connection != null) connection.close();
    if (preparedStatement != null) preparedStatement.close();
  } catch (SQLException e) {
    throw new StageException("Error closing database resources", e);
  }
}

What belongs in the constructor vs. start()

The constructor runs once when the stage is instantiated (before worker threads are assigned). start() runs once per worker thread just before processing begins. In practice:

• Constructor: Read config values into instance fields. Do not open connections or load large resources here — the constructor is called during pipeline validation and startup, before workers are ready.
• start(): Open connections, load resources, compile expressions, validate runtime constraints.
• stop(): Release everything acquired in start().
• processDocument(): Use the resources prepared in start(). Do not open or close connections per document.

4 - Connectors

For conceptual documentation — what a Connector is, the lifecycle design, and how Connectors are decoupled from downstream components — see Architecture: Connector.

Configuring a Connector

To configure a Connector, provide its class and name in the config. Optionally specify the pipeline it feeds, a docIdPrefix for ID namespacing, and whether it requires document collapsing:

{
  name: "my-connector"
  class: "com.kmwllc.lucille.connector.FileConnector"
  pipeline: "my-pipeline"
  docIdPrefix: "files-"
  paths: ["/data/files"]
}

Common Parameters

These parameters are available on all Connectors via AbstractConnector:

Sequencing Multiple Connectors

A single Lucille run can chain multiple Connectors in sequence. Each Connector runs to completion (all its documents processed and indexed) before the next begins:

connectors: [
  { name: "parent-docs",  class: "...", pipeline: "pipeline1" },
  { name: "child-docs",   class: "...", pipeline: "pipeline1" }
]

Lucille Connectors (Core)

The following connectors are deprecated. Use FileConnector with a corresponding FileHandler instead.

Lucille Connectors (Plugins)

File Handler configuration (CSV, JSON, XML, custom) is documented on the File Connector page.

4.1 - File Connector

Source Code

The FileConnector traverses a file system and publishes a Lucille Document for each file it encounters. It supports local filesystems, Amazon S3, Azure Blob Storage, and Google Cloud Storage through a unified interface — a single connector config can traverse paths across multiple providers simultaneously. Optional File Handlers extract structured Documents from files that themselves contain data (CSV rows, JSON objects, XML elements).

Cloud Storage Configuration

When traversing cloud storage, provide authentication under the appropriate top-level config block alongside your connector config. Each provider also accepts an optional maxNumOfPages to limit how many file listings are loaded into memory per request.

Azure

azure {
  connectionString: "DefaultEndpointsProtocol=https;AccountName=..."
  # or:
  accountName: "myaccount"
  accountKey: "mykey"
  maxNumOfPages: 100
}

You must provide either connectionString, or both accountName and accountKey.

Google Cloud Storage

gcp {
  pathToServiceKey: "/path/to/service-account.json"
  maxNumOfPages: 100
}

Amazon S3

s3 {
  accessKeyId: "AKIA..."
  secretAccessKey: "..."
  region: "us-east-1"
  maxNumOfPages: 100
}

For S3 paths, percent-encode special characters in paths (e.g., s3://bucket/folder%20with%20spaces).

A single connector can traverse multiple paths across providers:

paths: ["file:///local/data", "s3://my-bucket/prefix", "az://container/path"]

The URI scheme selects the appropriate storage backend automatically. A local filesystem client is always available without additional configuration.

File Handlers

File Handlers process individual files and extract one or more Lucille Documents from their contents. Without a handler, the FileConnector publishes one document per file containing only the file metadata fields. With a handler, a CSV file becomes one document per row, a JSON file becomes one document per object, and so on.

File Handlers are configured under the fileHandlers block. The key (csv, json, xml) determines which handler applies to files with that extension. fileHandlers and fileOptions are two separate top-level keys in the connector config block — they are not nested inside each other.

fileHandlers: {
  csv {
    separatorChar: "|"
    docIdPrefix: "csv-"
  }
  json {}
  xml {
    chunkPath: "//record"
  }
}

All File Handlers support docIdPrefix (prepended to every generated Document ID).

CSV File Handler

com.kmwllc.lucille.core.fileHandler.CSVFileHandler

Extracts one Document per row from a CSV file.

fileHandlers: {
  csv {
    idField: "article_id"
    separatorChar: "|"
    filenameField: "source_file"
    lowercaseFields: true
    docIdPrefix: "article-"
  }
}

JSON File Handler

com.kmwllc.lucille.core.fileHandler.JSONFileHandler

Extracts one Document per JSON object. Supports both standard JSON files (a single object or array) and JSON Lines (.jsonl) format where each line is a separate JSON object.

fileHandlers: {
  json {
    idField: "doc_id"
    blacklist: ["internal_metadata", "_rev"]
    docIdPrefix: "doc-"
  }
}

XML File Handler

com.kmwllc.lucille.core.fileHandler.XMLFileHandler

Extracts Documents from an XML file by selecting elements matching an XPath expression. Each matched element becomes a Document; element child text content and attributes become Document fields.

fileHandlers: {
  xml {
    chunkPath: "//product"
  }
}

Video File Handler

Available via the lucille-video plugin (com.kmwllc:lucille-video). Requires FFmpeg to be installed on the system.

Extracts frames from video files as individual Lucille Documents. See the plugin documentation for configuration parameters and supported formats.

Custom File Handlers

To process a file format not covered by the built-in handlers:

1. Extend BaseFileHandler.
2. Declare public static final Spec SPEC = SpecBuilder.fileHandler()... (required).
3. Implement Iterator<Document> processFile(InputStream stream, String pathStr).

Reference the handler by its fully-qualified class name in the fileHandlers block. The class field can also override the built-in handler for a standard extension:

fileHandlers: {
  csv {
    class: "com.example.MyCustomCSVHandler"
    myCustomParam: "value"
  }
}

File Handlers accept an InputStream and return Documents via an Iterator. The InputStream and any underlying resources are closed when the Iterator’s hasNext() returns false. When working directly with File Handlers in code, always exhaust the returned Iterator.

File Options

File options control traversal behaviour — how the connector handles the files it finds.

When archive or compressed file handling is enabled, entries inside archives get composite paths using the ! separator:

s3://bucket/archive.zip!path/inside/archive/file.csv

Modification and publish cutoffs apply to both the container archive and its entries.

Filter Options

Filter options control which files are processed and published. All filter options are optional. When multiple options are specified, a file must satisfy all of them to be processed. Evaluation order:

1. File name must match at least one includes pattern (if any are specified).
2. File name must not match any excludes pattern.
3. File’s modification time must fall within lastModifiedCutoff (if specified).
4. File must not have been published within lastPublishedCutoff (if specified — requires state configuration).
5. In incremental publish mode, only new or modified files since the last run are published (requires state configuration).

Example using regex patterns and time-based filtering:

filterOptions: {
  includes: [".*\\.pdf$", ".*\\.docx$"]
  excludes: [".*\\.tmp$", ".*~$"]
  lastModifiedCutoff: "24h"
  lastPublishedCutoff: "7d"
  publishMode: "incremental"
}

Note on pathsToSkip: while you could create an excludes regex detailing directories you do not want to publish files for, there are performance benefits to using pathsToSkip instead. With this parameter, the FileConnector will not actually traverse the path and its contents. If you use excludes, the path will be traversed, but the individual documents are not processed & published.

Each entry may be a URI with a scheme (for example s3://bucket/path, gs://bucket/path, or file:///path/to/dir). As a convenience, local file system paths may also be provided without a scheme - either absolute (e.g. /path/to/dir) or relative to the working directory (e.g. path/to/dir) - and will be resolved to an absolute file:// path.

Incremental Mode and State

The FileConnector can persist state to a JDBC-compatible database to track which files have been published and when. State enables lastPublishedCutoff, incremental publish mode, and tombstone detection.

Configure a state database alongside the connector:

state {
  enabled: true                    # Set to false to disable state without removing the block
  driver: "org.h2.Driver"          # Default: embedded H2
  connectionString: "jdbc:h2:./lucille-state"
  jdbcUser: ""
  jdbcPassword: ""
  tableName: "file_state"          # Defaults to the connector name
  performDeletions: true
  pathLength: 200                  # Max length of the file path column
  runsBeforeExpiration: 1          # Consecutive absent runs before tombstone (default: 1)
}

If connectionString is omitted, an embedded H2 database is created at ./state/{CONNECTOR_NAME}.

A few constraints to be aware of when using state:

• Files that are moved or renamed will not have lastPublishedCutoff applied — their new path is not recognised as previously published.
• Similarly, remember that the files in a directory that is skipped (by pathsToSkip), since they are not traversed, are also not tracked at all during a stateful run.
• Capitalise directory names in paths consistently across runs. State lookups are case-sensitive.
• Each database table should be used by only one connector configuration. Sharing a table across connectors will corrupt state. This can happen in several ways: reusing the same explicit tableName across multiple connectors in the same config file; running two separate configs concurrently where both reference the same tableName; or omitting tableName in two different configs that happen to use the same connector name (since tableName defaults to the connector name). In all cases, two connectors writing to the same state table will overwrite each other’s records, leading to incorrect incremental behavior — files may be skipped or reprocessed unexpectedly.

Tombstone Generation

When filterOptions.sendTombstones: true is set (requires incremental mode and state), the connector detects files that existed in the previous run but are no longer present. For each deleted file it publishes a tombstone document — a document with file_expired: true and ___skipped: true. The skipped flag causes the tombstone to bypass all pipeline stages. Downstream, configure the Indexer to issue a delete when it encounters the marker field:

indexer {
  type: "solr"
  deletionMarkerField: "file_expired"
  deletionMarkerFieldValue: "true"
}

By default, a file is considered expired after being absent from a single run. You can increase this threshold with state.runsBeforeExpiration to provide a safety margin against transient storage issues (e.g., a temporary listing failure):

state {
  runsBeforeExpiration: 3   # File must be absent for 3 consecutive runs before tombstone
}

The value must be at least 1. When set to 1 (default), a file missing from one run immediately triggers a tombstone — the original behavior.

This parameter only applies in incremental mode.

Document Fields

Every document published by the FileConnector carries these fields:

When a File Handler (CSV, JSON, XML) processes a file, it produces child documents with their own fields rather than a single file-level document.

Implementation Notes

This section is for Component Developers implementing new connectors. Pipeline Authors can stop here.

StorageClient Pattern

The connector handles four storage backends through a StorageClient abstraction. During traversal, the appropriate client is selected by URI scheme:

private void traverseStoragePath(Publisher publisher, URI pathToTraverse) throws ConnectorException {
    String clientKey = pathToTraverse.getScheme() != null ? pathToTraverse.getScheme() : "file";
    StorageClient storageClient = storageClientMap.get(clientKey);
    // ...
    storageClient.traverse(publisher, params, stateManager);
}

StorageClient is an interface with implementations for each backend. Each implementation handles listing/paginating files, applying filter criteria, reading file content, publishing documents, and updating state. The createClients(config) factory method inspects the config for cloud provider blocks and instantiates the appropriate clients.

Lifecycle: execute → close

Initialisation of storage clients and the state manager is deferred to inside execute(), not the constructor. This avoids opening network connections before the connector actually runs:

@Override
public void execute(Publisher publisher) throws ConnectorException {
    initialize();  // Init storage clients + state manager

    for (URI resource : storageURIs) {
        traverseStoragePath(publisher, resource);
    }

    if (sendTombstones) {
        sendExpiredFileTombstones(publisher);
    }
}

@Override
public void close() {
    if (stateManager != null) stateManager.shutdown();
    for (StorageClient client : storageClientMap.values()) {
        client.shutdown();
    }
}

SPEC Declaration

The FileConnector SPEC demonstrates how to declare a complex, nested configuration:

public static final Spec SPEC = SpecBuilder.connector()
    .requiredList("paths", new TypeReference<List<String>>(){})
    .optionalParent(
        SpecBuilder.parent("filterOptions")
            .optionalList("includes", new TypeReference<List<String>>(){})
            .optionalList("excludes", new TypeReference<List<String>>(){})
            .optionalString("lastModifiedCutoff", "lastPublishedCutoff", "publishMode", "sendTombstones").build(),
        SpecBuilder.parent("fileOptions")
            .optionalBoolean("getFileContent", "handleArchivedFiles", "handleCompressedFiles")
            .optionalString("moveToAfterProcessing", "moveToErrorFolder").build(),
        SpecBuilder.parent("state")
            .optionalString("driver", "connectionString", "jdbcUser", "jdbcPassword", "tableName")
            .optionalBoolean("performDeletions")
            .optionalNumber("pathLength").build(),
        GCP_PARENT_SPEC,
        AZURE_PARENT_SPEC,
        S3_PARENT_SPEC)
    .optionalParent("fileHandlers", new TypeReference<Map<String, Map<String, Object>>>(){})
    .build();

Key patterns: SpecBuilder.connector() provides base properties; nested SpecBuilder.parent(...) blocks define each optional config section; cloud provider specs are defined as reusable constants; fileHandlers uses a TypeReference because its structure is dynamic (keys are file extensions).

Reusable Patterns for Other Complex Connectors

1. StorageClient pattern — abstract the data source behind an interface; select implementation by URI scheme or config key.
2. Deferred initialisation — don’t open connections in the constructor; do it in execute().
3. Constraint validation early — check for incompatible config combinations (e.g., multiple paths + moveToAfterProcessing) in the constructor so failures surface before any traversal begins.
4. Nested SPEC declarations — group related config into parent blocks; define cloud provider specs as reusable constants shared across connectors.
5. JDBC-backed state — make state optional; check config.hasPath("state") before constructing the state manager.
6. Filter pipeline — layer multiple filter criteria (regex, time-based, state-based) with a clear evaluation order.
7. Tombstone generation — detect deletions by comparing the current file set to the previous state snapshot; mark documents with a field the Indexer can act on.
8. FileHandler delegation — use pluggable handlers (with their own SPECs) for format-specific processing rather than embedding format logic in the connector.

4.2 - Database Connector

Source Code

The DatabaseConnector reads rows from any JDBC-compatible relational database and publishes each row as a Lucille Document. Column names become field names on the Document.

Basic Configuration

connectors: [
  {
    name: "db-connector"
    class: "com.kmwllc.lucille.connector.jdbc.DatabaseConnector"
    pipeline: "my-pipeline"

    driver: "org.postgresql.Driver"
    connectionString: "jdbc:postgresql://localhost:5432/mydb"
    jdbcUser: "username"
    jdbcPassword: ${?DB_PASSWORD}
    sql: "SELECT id, title, body, published_at FROM articles WHERE active = true"
    idField: "id"
  }
]

Configuration Parameters

Pre and Post SQL

Use preSQL and postSQL to run setup and teardown logic that must happen before and after the main query:

{
  name: "db-connector"
  class: "com.kmwllc.lucille.connector.jdbc.DatabaseConnector"
  pipeline: "my-pipeline"

  driver: "org.postgresql.Driver"
  connectionString: "jdbc:postgresql://localhost:5432/mydb"
  jdbcUser: ${?DB_USER}
  jdbcPassword: ${?DB_PASSWORD}

  preSQL: "CREATE TEMP TABLE export_snapshot AS SELECT * FROM articles WHERE active = true"
  sql: "SELECT id, title, body FROM export_snapshot ORDER BY id"
  postSQL: "DROP TABLE IF EXISTS export_snapshot"
  idField: "id"
}

postSQL runs only if preSQL and the main query both succeed. If the connector throws, postSQL is skipped but close() is always called.

Multi-Query Joins

otherSQLs allows you to enrich primary rows with data from additional queries at read time, without a SQL JOIN. Each secondary query must be ordered by its join key (which must be an integer column matching a column in the primary query).

{
  sql: "SELECT id, title FROM articles ORDER BY id"
  idField: "id"
  otherSQLs: ["SELECT article_id, tag_name FROM article_tags ORDER BY article_id"]
  otherJoinFields: ["article_id"]
}

For each primary row, the connector merges matching rows from otherSQLs as multi-valued fields onto the Document.

Incremental Ingest

The DatabaseConnector does not maintain internal state. For incremental ingest (only rows modified since the last run), filter at the SQL level:

SELECT id, title, updated_at FROM articles
WHERE updated_at > TIMESTAMP '2025-01-01 00:00:00'
ORDER BY updated_at ASC

Store the high-water mark externally (e.g., in a config file, environment variable, or the database itself) and substitute it via HOCON variable substitution.

MySQL Streaming

For large MySQL tables, set fetchSize to avoid loading the entire result set into memory:

{
  driver: "com.mysql.cj.jdbc.Driver"
  connectionString: "jdbc:mysql://localhost:3306/mydb?useCursorFetch=true"
  fetchSize: -2147483648
  sql: "SELECT id, title FROM large_table ORDER BY id"
  idField: "id"
}

Common JDBC Drivers

Integration with QueryDatabase Stage

For per-document database enrichment (joining a lookup table for each document mid-pipeline, rather than reading a source table), use the QueryDatabase Stage.

4.3 - Solr Connector

Source Code

The SolrConnector issues a query against a Solr collection and publishes each result as a Lucille Document. It is useful for cross-index enrichment pipelines — for example, reading documents from one Solr collection, enriching them, and indexing them into a different collection or backend.

Internally, the connector uses cursor-based pagination (cursorMark) to iterate through large result sets without loading the full result into memory. Results are sorted by idField ascending (required for cursor pagination).

Basic Configuration

connectors: [
  {
    name: "solr-source"
    class: "com.kmwllc.lucille.connector.SolrConnector"
    pipeline: "my-pipeline"

    solr {
      url: ["http://localhost:8983/solr"]
      useCloudClient: true
      defaultCollection: "source-collection"
    }

    solrParams {
      q: "*:*"
      fl: "id,title,body,author"
      fq: "status:active"
      rows: 100
    }

    idField: "id"
  }
]

Configuration Parameters

solr Connection Parameters

Query Parameters (solrParams)

The solrParams block accepts any standard Solr query parameters:

solrParams {
  q: "status:active AND type:article"
  fl: "id,title,body,published_date"
  fq: ["category:news", "language:en"]
  rows: 500
}

Common parameters:

• q — query string (default: *:* if omitted)
• fl — field list (fields to return; omit for all fields)
• fq — filter query (can be a single string or a list for multiple filters)
• rows — page size for cursor pagination (default: Solr’s default, typically 10)

The connector always adds sort: {idField} asc automatically (required for cursor pagination) and manages cursorMark internally.

Pre and Post Actions

preActions and postActions are lists of Solr update requests sent before and after the main query, respectively. They support the {runId} placeholder, which is replaced with the current run’s UUID at execution time.

The request format is controlled by useXml:

• useXml: false (default): requests are JSON strings
• useXml: true: requests are XML strings

Example: Mark documents as in-progress before reading, mark as done after

connectors: [
  {
    name: "solr-source"
    class: "com.kmwllc.lucille.connector.SolrConnector"
    pipeline: "process-pipeline"

    solr {
      url: ["http://localhost:8983/solr/my-collection"]
    }

    # JSON update — set run_id on all active documents before reading
    preActions: [
      "{\"add\":{\"doc\":{\"id\":\"marker\",\"run_id\":\"{runId}\"}}}"
    ]

    # JSON update — clean up after run
    postActions: [
      "{\"delete\":{\"query\":\"run_id:{runId}\"}}"
    ]

    solrParams {
      q: "status:active"
      fl: "id,title,body"
    }
  }
]

postActions run only if preActions and the main query both succeed. If the connector throws, postActions are skipped.

Cursor-Based Pagination

The connector iterates through results using Solr’s cursorMark mechanism, which avoids deep pagination performance issues. This requires:

1. Documents to be sorted by a unique field (the idField).
2. The idField to be indexed in Solr as a single-valued, non-analyzed field.

For large collections, tune rows in solrParams to balance memory use and round-trip count.

4.4 - Kafka Connector

Source Code

The KafkaConnector reads Documents from a Kafka topic and publishes them into a Lucille pipeline. This is distinct from Kafka’s role as the messaging layer in distributed mode — the KafkaConnector is a data source, reading documents produced by an upstream system.

Use Cases

• Streaming ingest from Kafka: An upstream application publishes documents (as JSON) to a Kafka topic, and Lucille reads them for enrichment and indexing.
• Connectorless distributed mode: In this deployment pattern, a third-party publisher puts documents onto a Kafka source topic, and Lucille Workers consume them directly. In this case the KafkaConnector is not used — Workers listen to the source topic directly.

Configuration

All Kafka connection parameters are nested under the kafka key within the connector config block.

connectors: [
  {
    name: "kafka-source"
    class: "com.kmwllc.lucille.connector.KafkaConnector"
    pipeline: "my-pipeline"

    kafka.bootstrapServers: "kafka1:9092,kafka2:9092"
    kafka.topic: "my-source-topic"
    kafka.consumerGroupId: "lucille-kafka-connector"
    kafka.clientId: "lucille-consumer-1"
    kafka.maxPollIntervalSecs: 600
    idField: "article_id"
    maxMessages: 10000
  }
]

Configuration Parameters

Message Format

The KafkaConnector expects each Kafka message value to be a JSON object. Each JSON object becomes a Lucille Document. Field names in the JSON map directly to Document field names.

Example Kafka message:

{
  "article_id": "art-001",
  "title": "Breaking News",
  "body": "Full article text...",
  "published_at": "2025-06-01T12:00:00Z"
}

Security

For Kafka clusters with TLS or SASL authentication, use the top-level kafka {} block (separate from the connector’s inline params) to provide properties files and security settings:

kafka {
  bootstrapServers: "kafka1:9092"
  securityProtocol: "SSL"
  consumerPropertyFile: "/path/to/consumer.properties"
  producerPropertyFile: "/path/to/producer.properties"
  adminPropertyFile: "/path/to/admin.properties"
}

securityProtocol, consumerPropertyFile, producerPropertyFile, and adminPropertyFile are properties of the top-level kafka {} block and apply to all Kafka communication in the process, not just the KafkaConnector.

Kafka as the Messaging Layer vs. as a Source

These are two distinct uses of Kafka in Lucille:

Both can be active simultaneously: a KafkaConnector reads data from one topic while Lucille’s distributed messaging uses separate internal topics.

Related

• Topology: Fully Distributed
• Topology: Connector-less Distributed
• Getting Started: Distributed Mode

4.5 - Parquet Connector

Source Code

The ParquetConnector reads Apache Parquet files — locally or from Amazon S3 — and publishes each row as a Lucille Document. Parquet is a columnar format commonly used to store pre-computed embeddings, feature vectors, and large datasets.

Maven dependency:

<dependency>
  <groupId>com.kmwllc</groupId>
  <artifactId>lucille-parquet</artifactId>
  <version>${lucille.version}</version>
</dependency>

Configuration

connectors: [
  {
    name: "parquet-source"
    class: "com.kmwllc.lucille.parquet.connector.ParquetConnector"
    pipeline: "my-pipeline"
    pathToStorage: "/data/embeddings.parquet"
    idField: "doc_id"
    fsUri: "file:///"
  }
]

Configuration Parameters

S3 Configuration

For S3, provide the filesystem URI and credentials:

connectors: [
  {
    name: "parquet-s3"
    class: "com.kmwllc.lucille.parquet.connector.ParquetConnector"
    pipeline: "my-pipeline"
    pathToStorage: "/prefix/embeddings"
    idField: "doc_id"
    fsUri: "s3a://my-bucket"
    s3Key: ${AWS_ACCESS_KEY_ID}
    s3Secret: ${AWS_SECRET_ACCESS_KEY}
  }
]

Notes

• The connector uses Hadoop’s filesystem abstraction (FileSystem) for path traversal. Unlike FileConnector, it does not use Lucille’s StorageClient infrastructure.
• Parquet files must have the .parquet extension to be processed.
• When paginating with start/limit, it is recommended to use individual Connectors for each Parquet file rather than a directory path.
• The Parquet format requires random-access reads (not sequential streaming), which is why it is implemented as a standalone Connector rather than a FileConnector FileHandler.

4.6 - RSS Connector

The RSSConnector

The RSSConnector allows you to publish Documents representing the items found in an RSS feed of your choice. Each Document will (optionally) contain fields from the RSS items, like the author, description, title, etc. By default, the Document IDs will be the item’s guid, which should be a unique identifier for the RSS item.

You can configure the RSSConnector to only publish recent RSS items, based on the pubDate found on the items. Also, it can run incrementally, refreshing the RSS feed after a certain amount of time until you manually stop it. The RSSConnector will avoid publishing Documents for the same RSS item more than once.

The Documents published may have any of the following fields, depending on how the RSS feed is structured:

• author (String)
• categories (List<String>)
• comments (List<String>)
• content (String)
• description (String)
• enclosures (List<JsonNode>). Each JsonNode contains:
  • type (String)
  • url (String)
  • May contain length (Long)
• guid (String)
• isPermaLink (Boolean)
• link (String)
• title (String)
• pubDate (Instant)

5 - Stages

For conceptual documentation — what a Stage is, the Stage contract, conditions as a design decision, and child document emission — see Architecture: Stage.

Configuring a Stage

To configure a Stage, provide its class in the config. You can also specify a name (for logging and error messages), conditions, and conditionPolicy:

{
  name: "AddRandomBoolean-First"
  class: "com.kmwllc.lucille.stage.AddRandomBoolean"
  field_name: "rand_bool_1"
  percent_true: 65
}

Each Stage also accepts its own implementation-specific parameters (like field_name and percent_true above). See the individual stage pages below for details.

Conditions

For any Stage, you can specify conditions in its config to control when the Stage processes a Document.

Condition parameters

values and valuesPath are mutually exclusive — specifying both is an error.

How matching works

With values or valuesPath: The condition passes if any of the listed fields contains any of the listed values. Matching is type-coerced to string — a boolean field true matches the value "true", an integer 10 matches "10". null is a valid value entry and will match a null field value.

Without values or valuesPath: The condition checks field existence only.

• operator: "must" — passes if all listed fields are present on the document.
• operator: "must_not" — passes if all listed fields are absent from the document.

conditionPolicy

When a stage has multiple conditions, conditionPolicy in the stage’s root config controls how they combine:

• "all" (default) — all conditions must be met
• "any" — at least one condition must be met

Examples

Run a stage only when a field exists:

{
  class: "com.kmwllc.lucille.stage.MyStage"
  conditions: [
    { fields: ["content"] }
  ]
}

Run a stage only when a field matches a value:

{
  name: "print-1"
  class: "com.kmwllc.lucille.stage.Print"
  conditions: [
    { fields: ["city"], values: ["Boston", "New York"] }
  ]
}

Skip a stage when a field is present (must_not existence check):

{
  class: "com.kmwllc.lucille.stage.OpenAIEmbed"
  conditions: [
    { fields: ["embedding"], operator: "must_not" }
  ]
}

Require multiple conditions (all must be met):

{
  class: "com.kmwllc.lucille.stage.OpenAIEmbed"
  conditionPolicy: "all"
  conditions: [
    { fields: ["content"] }
    { fields: ["content_type"], values: ["article"] }
  ]
}

Load match values from a file:

{
  class: "com.kmwllc.lucille.stage.DropDocument"
  conditions: [
    { fields: ["category"], valuesPath: "s3://my-bucket/excluded-categories.txt" }
  ]
}

For the full reference on controlling document fate and connector sequencing — conditions, skipping, dropping, error handling, child documents, and more — see Control Flow.

Stage Catalogue

See All Stages for a complete listing of all available stages organized by category, including their configuration parameters.

Detailed pages are available for more complex stages:

• ChunkText — Split long text fields into chunks for embedding and RAG pipelines.
• EmbeddedPython — Run Python code inside the JVM using GraalPy.
• ExternalPython — Delegate processing to an external Python process via Py4J.
• PromptOllama — Enrich documents using a locally-running LLM.
• QueryOpensearch — Execute OpenSearch search templates per document.

Plugin stages (TextExtractor, ApplyOCR, ApplyOpenNLPNameFinders, JlamaEmbed) are listed at the bottom of All Stages with their Maven dependencies.

5.1 - All Stages

This page lists all Stages available in lucille-core and as optional plugin modules.

All stages share common configuration parameters:

Field Manipulation

CopyFields

com.kmwllc.lucille.stage.CopyFields

Copies one or more source fields to destination fields.

{ class: "com.kmwllc.lucille.stage.CopyFields", source: ["title"], dest: ["title_copy"] }

RenameFields

com.kmwllc.lucille.stage.RenameFields

Renames fields by mapping old names to new names.

{ class: "com.kmwllc.lucille.stage.RenameFields", fieldMapping: { old_field: new_field } }

DeleteFields

com.kmwllc.lucille.stage.DeleteFields

Removes specified fields from a Document.

SetStaticValues

com.kmwllc.lucille.stage.SetStaticValues

Sets one or more fields to fixed, static values.

{ class: "com.kmwllc.lucille.stage.SetStaticValues", fields: { source: "my-connector", version: 2 } }

Concatenate

com.kmwllc.lucille.stage.Concatenate

Concatenates the values of multiple source fields into a destination field.

SplitFieldValues

com.kmwllc.lucille.stage.SplitFieldValues

Splits a field value (or multi-valued field) by a delimiter into a list.

RemoveDuplicateValues

com.kmwllc.lucille.stage.RemoveDuplicateValues

Removes duplicate values from multi-valued fields.

RemoveEmptyFields

com.kmwllc.lucille.stage.RemoveEmptyFields

Removes fields whose value is null, an empty string, or an empty list.

NormalizeFieldNames

com.kmwllc.lucille.stage.NormalizeFieldNames

Normalizes field names (e.g., lowercasing, replacing spaces with underscores).

DropValues

com.kmwllc.lucille.stage.DropValues

Removes specific values from multi-valued fields.

Text Processing

TrimWhitespace

com.kmwllc.lucille.stage.TrimWhitespace

Trims leading and trailing whitespace from string fields.

RemoveDiacritics

com.kmwllc.lucille.stage.RemoveDiacritics

Normalizes accented characters to their ASCII equivalents (e.g., é → e).

TruncateField

com.kmwllc.lucille.stage.TruncateField

Truncates string field values to a maximum length.

Length

com.kmwllc.lucille.stage.Length

Computes the length (number of characters or list elements) of field values and stores the result.

ApplyRegex

com.kmwllc.lucille.stage.ApplyRegex

Applies a regular expression to one or more fields. Can extract capture groups or check for matches.

ReplacePatterns

com.kmwllc.lucille.stage.ReplacePatterns

Performs pattern-based find-and-replace on string field values.

Base64Decode

com.kmwllc.lucille.stage.Base64Decode

Decodes a Base64-encoded field value into a byte array or string.

NormalizeText

com.kmwllc.lucille.stage.NormalizeText

Applies Unicode normalization and optional case folding to text fields.

ExtractFirstCharacter

com.kmwllc.lucille.stage.ExtractFirstCharacter

Extracts the first character of a string field.

CreateStaticTeaser

com.kmwllc.lucille.stage.CreateStaticTeaser

Generates a teaser (short excerpt) from a text field.

HashFieldValueToBucket

com.kmwllc.lucille.stage.HashFieldValueToBucket

Hashes a field’s value and assigns the document to a numbered bucket (useful for deterministic partitioning).

Type Conversion & Parsing

ParseDate

com.kmwllc.lucille.stage.ParseDate

Parses date strings using configurable format patterns and writes an Instant or formatted string.

ParseFloats

com.kmwllc.lucille.stage.ParseFloats

Parses a JSON array string (e.g., "[0.1, 0.2, 0.3]") into a list of floats.

ParseFilePath

com.kmwllc.lucille.stage.ParseFilePath

Extracts path components (directory, filename, extension) from a file path field.

ParseJson

com.kmwllc.lucille.stage.ParseJson

Parses a JSON string field into a JsonNode.

Timestamp

com.kmwllc.lucille.stage.Timestamp

Writes the current timestamp to a field.

Document Flow Control

DropDocument

com.kmwllc.lucille.stage.DropDocument

Marks a document as dropped. It will not be sent to the Indexer.

Typically used with conditions to selectively drop documents:

{
  class: "com.kmwllc.lucille.stage.DropDocument"
  conditions: [
    { fields: ["status"], values: ["deleted"] }
  ]
}

SkipDocument

com.kmwllc.lucille.stage.SkipDocument

Marks a document as skipped. It bypasses all downstream Stages but still reaches the Indexer. Used to issue deletes against a search backend.

{
  class: "com.kmwllc.lucille.stage.SkipDocument"
  conditions: [
    { fields: ["is_deleted"], values: ["true"] }
  ]
}

Contains

com.kmwllc.lucille.stage.Contains

Checks whether a field’s value is contained in a configured list. Sets a boolean result field.

EmitNestedChildren

com.kmwllc.lucille.stage.EmitNestedChildren

Extracts a nested array from a Document and emits each array element as an independent child Document.

CreateChildrenStage

com.kmwllc.lucille.stage.CreateChildrenStage

Generates child documents from the current document’s fields using configurable rules.

CollapseChildrenDocuments

com.kmwllc.lucille.stage.CollapseChildrenDocuments

Merges child document field values back onto the parent document.

HTML, XML, and Web

ApplyJSoup

com.kmwllc.lucille.stage.ApplyJSoup

Parses HTML content using JSoup and extracts text or attribute values using CSS selectors.

Each destination field definition requires:

• type: "text" (inner text) or "attr" (attribute value).
• selector: CSS selector.
• attr: (if type is "attr") the attribute name to extract.

{
  class: "com.kmwllc.lucille.stage.ApplyJSoup"
  byteArrayField: "html_content"
  destinationFields: {
    title: { type: "text", selector: "h1" }
    body:  { type: "text", selector: ".article-body p" }
  }
}

XPathExtractor

com.kmwllc.lucille.stage.XPathExtractor

Evaluates XPath expressions against an XML field.

FetchUri

com.kmwllc.lucille.stage.FetchUri

Fetches the content of a URL stored in a document field and stores the response as a byte array.

File Handling

ApplyFileHandlers

com.kmwllc.lucille.stage.ApplyFileHandlers

Applies configured FileHandlers to a byte array field, generating child documents for each extracted record.

FetchFileContent

com.kmwllc.lucille.stage.FetchFileContent

Loads the content of a file path (from local filesystem or cloud storage) into a byte array field on the document.

ComputeFieldSize

com.kmwllc.lucille.stage.ComputeFieldSize

Measures the byte size of a field value (e.g., a byte array or string) and stores the result.

TextExtractor

com.kmwllc.lucille.tika.stage.TextExtractor (requires lucille-tika)

Extracts text from over 1,000 file formats — PDF, Microsoft Office documents, HTML, images with embedded OCR, and many more — using Apache Tika. Reads raw bytes from a source field and writes extracted text to a destination field.

Maven dependency:

<dependency>
  <groupId>com.kmwllc</groupId>
  <artifactId>lucille-tika</artifactId>
  <version>${lucille.version}</version>
</dependency>

{
  name: "extract-text"
  class: "com.kmwllc.lucille.tika.stage.TextExtractor"
  source: "file_content"
  dest: "extracted_text"
}

ApplyOCR

com.kmwllc.lucille.ocr.stage.ApplyOCR (requires lucille-ocr)

Performs optical character recognition on image fields using Tesseract. Reads image bytes from a source field and writes the recognized text to a destination field. Requires Tesseract to be installed on the system running Lucille.

Maven dependency:

<dependency>
  <groupId>com.kmwllc</groupId>
  <artifactId>lucille-ocr</artifactId>
  <version>${lucille.version}</version>
</dependency>

{
  name: "ocr"
  class: "com.kmwllc.lucille.ocr.stage.ApplyOCR"
  source: "image_bytes"
  dest: "ocr_text"
  language: "eng"
}

Enrichment & Lookup

DictionaryLookup

com.kmwllc.lucille.stage.DictionaryLookup

Looks up field values in a term dictionary and adds matched entries as field values.

QueryDatabase

com.kmwllc.lucille.stage.QueryDatabase

Executes a JDBC prepared statement using document field values as parameters and merges the result onto the document. Useful for per-document database enrichment (e.g., joining a lookup table for each document mid-pipeline).

ElasticsearchLookup

com.kmwllc.lucille.stage.ElasticsearchLookup

Performs a lookup against an Elasticsearch index and merges matching fields onto the document.

QueryOpensearch

com.kmwllc.lucille.stage.QueryOpensearch

Executes a search template against an OpenSearch index using document field values as parameters. See QueryOpensearch for full documentation.

MatchQuery

com.kmwllc.lucille.stage.MatchQuery

Executes a match query against a configured search backend and enriches the document with matching results.

AI / ML

OpenAIEmbed

com.kmwllc.lucille.stage.OpenAIEmbed

Generates vector embeddings for a text field using the OpenAI Embeddings API.

Supported models: text-embedding-3-small, text-embedding-3-large, text-embedding-ada-002

Text is truncated to 8,191 tokens before embedding (the OpenAI API limit). Lucille uses jtokkit for accurate token counting before the API call.

{
  class: "com.kmwllc.lucille.stage.OpenAIEmbed"
  source: "content"
  dest: "content_vector"
  modelName: "text-embedding-3-small"
  apiKey: ${OPENAI_API_KEY}
  embedDocument: true
  embedChildren: true
}

JlamaEmbed

com.kmwllc.lucille.jlama.stage.JlamaEmbed (requires lucille-jlama)

Generates vector embeddings using a quantized LLM running locally inside the JVM via Jlama. No API key or external service required — the model runs directly in the Lucille process. Useful for teams with data-residency or compliance constraints that prevent sending documents to external APIs.

Maven dependency:

<dependency>
  <groupId>com.kmwllc</groupId>
  <artifactId>lucille-jlama</artifactId>
  <version>${lucille.version}</version>
</dependency>

{
  name: "embed"
  class: "com.kmwllc.lucille.jlama.stage.JlamaEmbed"
  source: "content"
  dest: "content_vector"
  modelPath: "/models/my-embedding-model"
}

PromptOllama

com.kmwllc.lucille.stage.PromptOllama

Sends document fields to a locally-running Ollama LLM and merges the JSON response back onto the document. See PromptOllama for full documentation.

EmbeddedPython

com.kmwllc.lucille.stage.EmbeddedPython

Runs per-document Python code inside the JVM using GraalPy. See EmbeddedPython for full documentation.

ExternalPython

com.kmwllc.lucille.stage.ExternalPython

Delegates per-document processing to an external Python process via Py4J. See ExternalPython for full documentation.

ApplyJavascript

com.kmwllc.lucille.stage.ApplyJavascript

Runs a JavaScript snippet per document using GraalVM’s JavaScript engine.

ApplyJSONata

com.kmwllc.lucille.stage.ApplyJSONata

Applies a JSONata expression to transform the document’s JSON representation.

ExtractEntitiesFST

com.kmwllc.lucille.stage.ExtractEntitiesFST

Performs named entity recognition using a finite-state transducer dictionary.

ExtractEntities

com.kmwllc.lucille.stage.ExtractEntities

Extracts entities from text fields using configured rules.

ApplyOpenNLPNameFinders (requires lucille-entity-extraction)

com.kmwllc.lucille.entity.stage.ApplyOpenNLPNameFinders

Performs named entity recognition (NER) using Apache OpenNLP models. Identifies entities such as people, organizations, and locations in text fields.

Maven dependency:

<dependency>
  <groupId>com.kmwllc</groupId>
  <artifactId>lucille-entity-extraction</artifactId>
  <version>${lucille.version}</version>
</dependency>

DetectLanguage

com.kmwllc.lucille.stage.DetectLanguage

Detects the language of a text field and writes the ISO language code to a destination field.

ChunkText

com.kmwllc.lucille.stage.ChunkText

Splits a long text field into smaller chunks suitable for embedding (RAG pipelines). Each chunk is emitted as a child document. See ChunkText for full documentation.

Chunking methods:

• sentence — Detects sentence boundaries using OpenNLP.
• paragraph — Splits on consecutive line breaks (\n\n, \r\n\r\n, etc.).
• fixed — Splits every lengthToSplit characters.
• custom — Splits on occurrences of the regex pattern.

Child document fields: Each child document receives id (parent ID + chunk number), parent_id, offset, length, chunk_number, total_chunks, and the chunk content in dest.

RandomVector

com.kmwllc.lucille.stage.RandomVector

Generates a random float vector and sets it on a document field. Useful for testing vector search pipelines.

Testing & Debugging

Print

com.kmwllc.lucille.stage.Print

Logs documents in JSON format at INFO level and/or writes them to a file. Place Print anywhere in the pipeline to capture the document state at that point.

{
  class: "com.kmwllc.lucille.stage.Print"
  shouldLog: false
  outputFile: "/tmp/pipeline-output.jsonl"
}

Capture and Replay

Print enables a useful development pattern: run a pipeline to capture its output to disk, then replay that output directly into a search backend without re-running the enrichment.

Step 1 — Capture. Add Print to the end of your pipeline with an outputFile. Use a NopIndexer (or set sendEnabled: false on a real indexer) so no data is actually indexed during the capture run:

pipelines: [{
  name: my-pipeline
  stages: [
    { class: "com.kmwllc.lucille.stage.SomeExpensiveEnrichment" ... }
    {
      class: "com.kmwllc.lucille.stage.Print"
      shouldLog: false
      outputFile: "/tmp/captured.jsonl"
    }
  ]
}]
indexer { type: Nop }

Step 2 — Replay. Point a FileConnector at the captured JSONL file using the JSON file handler. Use a minimal or empty pipeline — the captured documents already contain all enriched fields:

connectors: [{
  name: replay
  class: "com.kmwllc.lucille.connector.FileConnector"
  pipeline: replay-pipeline
  paths: ["/tmp/captured.jsonl"]
  fileHandlers: { json: { idField: "id" } }
}]
pipelines: [{
  name: replay-pipeline
  stages: []
}]
indexer { type: OpenSearch }
opensearch { ... }

This lets you iterate on indexer configuration, field mappings, or search backend settings without repeating expensive enrichment (OCR, embedding generation, database lookups) on every attempt.

AddRandomString

com.kmwllc.lucille.stage.AddRandomString

Adds a random alphanumeric string to a field.

AddRandomInt

com.kmwllc.lucille.stage.AddRandomInt

Adds a random integer to a field.

AddRandomDouble

com.kmwllc.lucille.stage.AddRandomDouble

Adds a random double to a field.

AddRandomDate

com.kmwllc.lucille.stage.AddRandomDate

Adds a random date/timestamp to a field within a configured range.

AddRandomBoolean

com.kmwllc.lucille.stage.AddRandomBoolean

Adds a random boolean to a field.

AddRandomNestedField

com.kmwllc.lucille.stage.AddRandomNestedField

Adds a nested JSON object with random values to a field. Useful for testing nested document structures.

5.2 - ChunkText

The ChunkText Stage splits a long text field into smaller, optionally overlapping segments. Each segment is emitted as a child document that flows independently through all downstream stages and is indexed as its own record. This is the foundation of retrieval-augmented generation (RAG) pipelines in Lucille.

Configuration

{
  class: "com.kmwllc.lucille.stage.ChunkText"
  source: "body"
  dest: "text"
  chunkingMethod: "sentence"
  chunksToMerge: 5
  chunksToOverlap: 1
  cleanChunks: true
}

Configuration Parameters

Chunking Methods

sentence (default)

Detects sentence boundaries using an Apache OpenNLP sentence detector. This produces semantically coherent chunks. Best used with chunksToMerge to combine a few sentences into each final chunk.

{
  class: "com.kmwllc.lucille.stage.ChunkText"
  source: "body"
  chunkingMethod: "sentence"
  chunksToMerge: 4      # 4 sentences per chunk
  chunksToOverlap: 1    # 1 sentence of overlap between chunks
}

paragraph

Splits on consecutive line breaks (\n\n, \r\n\r\n, etc.). Suitable for documents with paragraph structure.

{
  class: "com.kmwllc.lucille.stage.ChunkText"
  source: "body"
  chunkingMethod: "paragraph"
  characterLimit: 1000    # Cap chunks at 1000 characters
}

fixed

Splits every lengthToSplit characters. Simple and predictable, but may cut mid-sentence.

{
  class: "com.kmwllc.lucille.stage.ChunkText"
  source: "body"
  chunkingMethod: "fixed"
  lengthToSplit: 512
}

custom

Splits on occurrences of a regex pattern. Useful when documents have a known structural delimiter.

{
  class: "com.kmwllc.lucille.stage.ChunkText"
  source: "body"
  chunkingMethod: "custom"
  regex: "---+"    # Split on markdown horizontal rules
}

Processing Order

The Stage applies transformations in this order:

1. Initial chunking by the chosen method.
2. Cleaning (if cleanChunks: true): strip newlines, trim whitespace.
3. Pre-merge filtering: drop or truncate chunks based on preMergeMinChunkLen / preMergeMaxChunkLen.
4. Merging (if chunksToMerge > 1): combine N chunks into each final chunk.
5. Overlap (if chunksToOverlap or overlapPercentage set): prepend content from adjacent chunks.
6. Character limiting (if characterLimit set): truncate oversized final chunks.

Child Document Fields

Each chunk becomes a child document with these fields:

All other fields from the parent document are not copied to child documents unless a downstream Stage does so explicitly.

Typical RAG Pipeline

pipelines: [
  {
    name: "rag-pipeline"
    stages: [
      # 1. Extract text from PDF bytes
      {
        class: "com.kmwllc.lucille.tika.stage.TextExtractor"
        source: "file_content"
        dest: "body"
      },
      # 2. Split into sentence-merged chunks
      {
        class: "com.kmwllc.lucille.stage.ChunkText"
        source: "body"
        dest: "chunk_text"
        chunkingMethod: "sentence"
        chunksToMerge: 5
        chunksToOverlap: 1
        cleanChunks: true
        characterLimit: 2000
      },
      # 3. Embed each chunk (only child docs, which carry the chunk text)
      {
        class: "com.kmwllc.lucille.stage.OpenAIEmbed"
        source: "chunk_text"
        dest: "chunk_vector"
        embedDocument: false
        embedChildren: true
        apiKey: ${OPENAI_API_KEY}
      }
    ]
  }
]

Tips

• Use embedDocument: false, embedChildren: true in OpenAIEmbed to only embed chunks, not the full parent document.
• Set characterLimit to stay within your embedding model’s token limit. For text-embedding-3-small, 8,191 tokens ≈ roughly 6,000–7,000 characters of English text.
• Use cleanChunks: true when the source text has formatting artifacts (extra whitespace, embedded newlines from PDF extraction).
• chunksToOverlap and chunksToMerge work together: if you merge 5 sentences with 1 overlap, each chunk shares its last sentence with the next chunk, helping the model retrieve context that spans a chunk boundary.

5.3 - PromptOllama

What if you could just, actually, put an LLM on everything?

Ollama

Ollama allows you to run a variety of Large Language Models (LLMs) with minimal setup. You can also create custom models using Modelfiles and system prompts.

The PromptOllama Stage allows you to connect to a running instance of Ollama Server, which communicates with an LLM through a simple API. The Stage sends part (or all) of a Document to the LLM for generic enrichment. You’ll want to create a custom model (with a Modelfile) or provide a System Prompt in the Stage Config that is tailored to your pipeline.

We strongly recommend you have the LLM output only a JSON object for two main reasons: Firstly, LLMs tend to follow instructions better when instructed to do so. Secondly, Lucille can then parse the JSON response and fully integrate it into your Document.

Example

Let’s say you are working with Documents which represent emails, and you want to monitor them for potential signs of fraud. Lucille doesn’t have a DetectFraud Stage (at time of writing), but you can use PromptOllama to add this information with an LLM.

• Modelfile: Let’s say you created a custom model, fraud_detector, in your instance of Ollama Server. As part of the modelfile, you instruct the model to check the contents for fraud and output a JSON object containing just a boolean value (under fraud). Your Stage would be configured like so:

{
  name: "Ollama-Fraud"
  class: "com.kmwllc.lucille.stage.PromptOllama"
  hostURL: "http://localhost:9200"
  modelName: "fraud_detector"
  fields: ["email_text"]
}

• System Prompt: You can also just reference a specific LLM directly, and provide a system prompt in the Stage configuration.

{
  name: "Ollama-Fraud"
  class: "com.kmwllc.lucille.stage.PromptOllama"
  hostURL: "http://localhost:9200"
  modelName: "gemma3"
  systemPrompt: """
    You are to read the text inside "email_text" and output a JSON object
    containing only one field, fraud, a boolean, representing whether the
    text contains evidence of fraud or not.
  """
  fields: "email_text"
}

Regardless of the approach you choose, the LLM will receive a request that looks like this:

{
  "email_text": "Let's be sure to juice the numbers in our next quarterly earnings report."
}

(Since fields: ["email_text"], any other fields on this Document are not part of the request.)

And the response from the LLM should look like this:

{
  "fraud": true
}

Lucille will then add all key-value pairs in this response JSON into your Document. So, the Document will become:

{
  "id": "emails.csv-85",
  "run-id": "f9538992-5900-459a-90ce-2e8e1a85695c",
  "email_text": "Let's be sure to juice the numbers in our next quarterly earnings report.",
  "fraud": true
}

As you can see, PromptOllama is very versatile, and can be used to enrich your Documents in a lot of ways.

5.4 - QueryOpensearch

com.kmwllc.lucille.stage.QueryOpensearch

OpenSearch Templates

You can use templates in OpenSearch to repeatedly run a certain query using different parameters. For example, if we have an index full of parks, and we want to search for a certain park, we might use a template like this:

{
  "source": {
    "query": {
      "match_phrase": {
        "park_name": "{{park_to_search}}"
      }
    }
  }
}

In Opensearch, you could then call this template (providing it park_to_search) instead of writing out the full query each time you want to search.

Templates can also have default values. For example, if you want park_to_search to default to “Central Park” when a value is not provided, it would be written as: "park_name": "{{park_to_search}}{{^park_to_search}}Central Park{{/park_to_search}}"

QueryOpensearch Stage

The QueryOpensearch Stage executes a search template using certain fields from a Document as your parameters and adding OpenSearch’s response to the Document. You’ll specify either templateName, the name of a search template you’ve saved, or searchTemplate, the template you want to execute, in your Config.

You’ll also need to specify the names of parameters in your search template. These will need to match the names of fields on your Documents. If your names don’t match, you can use the RenameFields Stage first.

In particular, you have to specify which parameters are required and which are optional. If a required name in requiredParamNames is missing from a Document, an Exception will be thrown, and the template will not be executed. If an optional name in optionalParamNames is missing they (naturally) won’t be part of the template execution, so the default value will be used by OpenSearch.

If a parameter without a default value is missing, OpenSearch doesn’t throw an Exception - it just returns an empty response with zero hits. So, it is very important that requiredParamNames and optionalParamNames are defined very carefully!

Connection and Authentication

The stage connects to OpenSearch via the same opensearch config block used by the OpenSearch Indexer. Authentication and TLS are handled by OpenSearchUtils:

• Basic auth: Embed credentials in the URL: https://username:password@host:9200
• Self-signed certs: Set acceptInvalidCert: true to skip certificate validation (development only).

{
  name: "QueryOpensearch-Example"
  class: "com.kmwllc.lucille.stage.QueryOpensearch"

  opensearch {
    url: "https://admin:admin@localhost:9200"
    index: "my-index"
    acceptInvalidCert: true
  }

  templateName: "my-saved-template"
  requiredParamNames: ["park_to_search"]
  optionalParamNames: ["city"]
  destinationField: "search_results"
  opensearchResponsePath: "/hits/hits"
}

Configuration Parameters

5.5 - EmbeddedPython

Why Use It?

EmbeddedPython executes per-document Python code inside the Lucille JVM using GraalPy. Instead of returning a JSON object, your script mutates the current document directly through a Python-friendly proxy bound as doc (and the raw Java document as rawDoc). This avoids ports, subprocesses, venvs, and per-document JSON round trips.

When To Use It

Use EmbeddedPython when you need one or more of the following:

• Minimal operational overhead (ports, subprocess lifecycle, venv creation, pip installs).
• No use of any external Python libraries or native dependencies that require a real Python environment.
• Lightweight field enrichment/transformation.

When To Use ExternalPython Instead

Avoid EmbeddedPython and use ExternalPython when you need one or more of the following:

• Real Python compatibility (including packages with native dependencies).
• Dependency management via a requirements.txt installed into a managed venv.
• Process isolation apart from the JVM.

Example

Input Document

{
  "id": "doc-1",
  "title": "Hello",
  "author": "Test",
  "views": 123
}

Python Script

doc["title"] = doc["title"].upper()

Output Document

{
  "id": "doc-1",
  "title": "HELLO",
  "author": "Test",
  "views": 123
}

Config Parameters

{
 name: "EmbeddedPython-Example"
 class: "com.kmwllc.lucille.stage.EmbeddedPython"

 # Specify exactly one of the following:
 script_path: "/path/to/my_script.py"
 script: "doc['title'] = doc['title'].upper()"
}

script_path can begin with classpath: to load the script as a classpath resource rather than a filesystem path:

script_path: "classpath:scripts/my_script.py"

This lets you package Python scripts inside your JAR as resources (under src/main/resources/). The Lucille process loads them directly from the classpath at runtime, so you don’t need to place files in specific filesystem locations or ensure the process has read access to them.

5.6 - ExternalPython

Why Use It?

ExternalPython delegates per-document processing to an external Python process using Py4J. Lucille serializes the Document into a request, calls a Python function, receives a JSON response, and applies that response back onto the document.

When To Use It

Use ExternalPython when you need one or more of the following:

• Real Python compatibility (including packages with native dependencies).
• Dependency management via a requirements.txt installed into a managed venv.
• Process isolation apart from the JVM.

When To Use EmbeddedPython Instead

Avoid ExternalPython and use EmbeddedPython when you need one or more of the following:

• Minimal operational overhead (ports, subprocess lifecycle, venv creation, pip installs).
• No use of any external Python libraries or native dependencies that require a real Python environment.
• Lightweight field enrichment/transformation.

Restrictions

Your python file must be in one of the following directories that start in the current working directory that is running lucille:

• ./python
• ./src/main/resources
• ./src/test/resources
• ./src/test/resources/ExternalPythonTest (for testing)

Example

Input Document

{
  "id": "doc-1",
  "title": "Hello",
  "author": "Test",
  "views": 123
}

Python Script

def process_document(doc):
    title = doc["title"]
  
    return {
        "title": title.upper()
    }

Python Returns

{
  "title": "HELLO"
}

Output Document

{
  "id": "doc-1",
  "title": "HELLO"
}

Config Parameters

{
  name: "ExternalPython-Example"
  class: "com.kmwllc.lucille.stage.ExternalPython"

  scriptPath: "/path/to/my_script.py"

  # Optional
  pythonExecutable: "python3"
  requirementsPath: "/path/to/requirements.txt"
  functionName: "process_document"
  port: 25333
}

Example (NumPy)

Input Document

{
  "id": "doc-2",
  "values": [1, 2, 3, 4, 5]
}

Python Script

import numpy as np

def process_document(doc):
    arr = np.array(doc["values"], dtype=float)
  
    return {
        "values": doc["values"],
        "mean": float(np.mean(arr)),
        "stddev": float(np.std(arr))
    }

Output Document

{
  "id": "doc-2",
  "values": [1, 2, 3, 4, 5],
  "mean": 3.0,
  "stddev": 1.41
}

requirements.txt

numpy

Config Parameters

{
  name: "ExternalPython-Numpy"
  class: "com.kmwllc.lucille.stage.ExternalPython"

  scriptPath: "/path/to/my_numpy_script.py"
  requirementsPath: "/path/to/requirements.txt"
  
  # Optional
  pythonExecutable: "python3"
  functionName: "process_document"
  port: 25333
}

6 - Indexers

For conceptual documentation — what an Indexer is, why batching matters, deletion as a design pattern, and error handling at the batch level — see Architecture: Indexer.

Generic indexer Configuration

Indexer configuration has two parts: the generic indexer block (common to all backends), and a backend-specific config block (e.g., solr, opensearch, elastic, csv).

indexer {
  type: "Solr"
  batchSize: 100
  batchTimeout: 100
  blacklist: ["internal_field"]
}

solr {
  url: "http://localhost:8983/solr/my-collection"
}

indexer.type is shorthand for a built-in indexer: "Solr", "OpenSearch", "Elasticsearch", or "CSV". For plugin indexers, use indexer.class with the fully qualified class name instead.

Generic Parameters

Field Filtering

Whitelist and blacklist are applied at indexing time, not during pipeline processing. Stages see all fields on a document; only the Indexer strips fields before sending to the backend. Reserved internal fields (___dropped, ___skipped, ___children) are always stripped.

When indexOverrideField is set, the Indexer uses a MultiBatch — maintaining a separate batch per distinct field value and flushing each independently when it reaches batchSize or batchTimeout.

Deletion Mechanics

Two distinct deletion mechanisms are available:

Delete by ID: Set deletionMarkerField and deletionMarkerFieldValue. When a document has the marker field set to the marker value, the Indexer issues a delete-by-ID against the search backend for that document’s ID.

Delete by query: Also set deleteByFieldField and deleteByFieldValue. When a document has all four deletion fields set, the Indexer issues a delete-by-query: it deletes all documents in the index where deleteByFieldField’s referenced field equals the value in deleteByFieldValue’s referenced field.

indexer {
  type: "Solr"
  deletionMarkerField: "file_expired"
  deletionMarkerFieldValue: "true"
  deleteByFieldField: "delete_by_field"
  deleteByFieldValue: "delete_by_value"
}

Batching Behavior

Documents accumulate in a batch and are flushed when any of these conditions is met:

• The batch reaches batchSize documents (default: 100).
• The batch reaches batchByteSize bytes of accumulated document payload (disabled by default).
• batchTimeout milliseconds have elapsed since the last document was added or the last flush (default: 100ms).

When only batchByteSize is set (without batchSize), document-count batching is effectively disabled — batches flush purely on payload size. When both are set, whichever limit is reached first triggers the flush. This is useful for backends with request-size limits (e.g., a 10 MB bulk API limit) where a fixed document count may produce unpredictably sized payloads.

The timeout flush ensures documents are not left waiting indefinitely in low-volume scenarios.

Batch-level vs. per-document failures: A bulk-API failure that rejects the entire request fails all documents in the batch. Individual document rejections in the response (e.g., mapping errors) fail only those specific documents — the rest succeed. Both cases are tracked separately in the run summary.

Indexer Catalogue

Core Indexers

• Solr Indexer — Single-node and SolrCloud.
• OpenSearch Indexer — OpenSearch with optional partial updates.
• Elasticsearch Indexer — Elasticsearch with join field support.
• CSV Indexer — Write pipeline output to a CSV file.
• NopIndexer — No-op indexer for testing.

Plugin Indexers

• Pinecone Indexer — Index vector embeddings into Pinecone.
• Weaviate Indexer — Index documents and vectors into Weaviate.

6.1 - Solr Indexer

com.kmwllc.lucille.indexer.SolrIndexer

Config block: solr { ... }

*Provide either url or (for SolrCloud) zkHosts.

# Single node
solr { url: "http://localhost:8983/solr/my-collection" }

# SolrCloud with URL
solr {
  useCloudClient: true
  url: ["http://localhost:8983/solr"]
  defaultCollection: "my-collection"
}

# SolrCloud with ZooKeeper
solr {
  useCloudClient: true
  zkHosts: ["zk1:2181", "zk2:2181"]
  zkChroot: "/solr"
  defaultCollection: "my-collection"
}

Child Document Support

SolrIndexer is the only Lucille indexer that fully supports attached child documents. When a document has children (via doc.addChild()), they are automatically converted to Solr’s _childDocuments_ structure. The ___children field itself is never sent to Solr — only the nested Solr child documents are added.

Children respect the configured blacklist and whitelist: filtered fields are removed from child documents before they are added to the Solr parent.

Collection Routing

When indexer.indexOverrideField is set, documents are routed to different Solr collections within the same batch. The value of the override field on each document determines which collection receives it. The override field itself is stripped from the document before sending.

The target collection must already exist in Solr — the indexer does not create collections.

Interleaved Add/Delete Ordering

SolrIndexer preserves the order of adds and deletes within a batch. If a batch contains an add followed by a delete for the same document ID (or vice versa), the operations are sent to Solr in the correct sequence. This is important for incremental ingestion where a document may be re-added and then deleted (or deleted and re-added) within the same run.

Internally, the indexer detects when an add and a delete for the same ID appear in the same batch and flushes the earlier operation before queuing the later one.

Delete-by-Query via Terms Queries

When deleteByFieldField and deleteByFieldValue are configured, SolrIndexer constructs a Solr terms query for efficient bulk deletion rather than issuing individual delete-by-query calls. Multiple deletions targeting the same field are combined into a single terms query:

(+{!terms f='category' v='obsolete,deprecated'})

This reduces the number of round-trips to Solr when many documents are marked for deletion in the same batch.

Limitations

• Object fields not supported — If a document field contains a nested Map or Object, SolrIndexer throws an IndexerException. Flatten nested structures in your pipeline before indexing to Solr.

Troubleshooting

Connection validation fails:

• Single-node mode (useCloudClient: false): Validation uses solrClient.ping(). If this fails, check that the URL includes the collection path (e.g., http://localhost:8983/solr/my-collection) and that the collection exists.
• SolrCloud mode (useCloudClient: true): Validation checks cluster status via the Collections API. If this fails, the cluster is unreachable — check ZooKeeper connectivity and that at least one Solr node is running.

“Object field is not supported by the SolrIndexer”: A document field contains a nested JSON object. Add a stage to your pipeline that flattens or removes the field before it reaches the indexer.

6.2 - OpenSearch Indexer

com.kmwllc.lucille.indexer.OpenSearchIndexer

Config block: opensearch { ... }

Also supports indexer.routingField and indexer.versionType (via the generic indexer block).

opensearch {
  url: "https://admin:admin@localhost:9200"
  url: ${?OPENSEARCH_URL}
  index: "my-index"
  acceptInvalidCert: true
}

Routing

When indexer.routingField is set, the value of that field on each document is used as the _routing parameter in the bulk request. This controls which shard receives the document. Use this when your OpenSearch index has custom shard routing configured.

Version Type

When indexer.versionType is set to external or external_gte, the Kafka message offset is used as the document’s version number. This enables optimistic concurrency control in streaming mode — OpenSearch will reject a write if the incoming version is not greater than (or equal to, for external_gte) the existing version.

By default this uses the Kafka offset (requiring distributed mode with KafkaDocument instances). To use a version value from the document itself — useful in local mode or when the version is sourced from an upstream system — set indexer.versionField:

indexer {
  versionType: "external_gte"
  versionField: "my_version_field"
}

If versionField is set but the field is absent on a particular document, and the document is not a KafkaDocument, the version is omitted and the document is indexed without optimistic concurrency control.

Partial Update Mode

When update: true, documents are sent as partial updates (doc-as-upsert) rather than full index operations:

• Full index (default): Replaces the entire document in the index. Fields not present in the new document are removed.
• Partial update: Merges fields into the existing document. Fields not present in the update are left unchanged.

opensearch {
  url: ${OPENSEARCH_URL}
  index: "my-index"
  update: true
}

Index Override

OpenSearch supports indexer.indexOverrideField for routing documents to different indices within the same batch. The value of the override field on each document determines the target index. The override field itself is stripped from the document before sending.

Retry Behavior

OpenSearchIndexer wraps both transport-level failures (connection refused, timeout) and HTTP-level failures (e.g., 429, 503) as IndexerRetryableException. This enables the base class retry machinery when indexer.maxRetries is configured.

Per-document failures in the bulk response are also wrapped with the item’s HTTP status code, allowing the retry logic to distinguish between retryable failures (e.g., 429 Too Many Requests) and permanent failures (e.g., 400 Bad Request).

Child Documents

Attached children are flattened into the parent document as nested objects. They are not indexed as separate OpenSearch documents. This is different from Solr’s _childDocuments_ approach — in OpenSearch, children become part of the parent’s JSON structure.

Troubleshooting

“index not found”: The target index must exist before indexing. Create it manually or via an index template.

TLS certificate errors: Set acceptInvalidCert: true for development environments with self-signed certificates. Do not use this in production.

Authentication: Include credentials in the URL: https://user:password@host:9200. Use environment variable substitution to avoid hardcoding: url: ${OPENSEARCH_URL}.

6.3 - Elasticsearch Indexer

com.kmwllc.lucille.indexer.ElasticsearchIndexer

Config block: elastic { ... }

Supports all OpenSearch parameters, plus parent-child join support:

Join field support (for parent-child mappings):

elastic {
  url: "http://localhost:9200"
  index: "my-index"
  join: {
    joinFieldName: "my_join_field"
    isChild: true
    childName: "my_child"
    parentDocumentIdSource: "parent_id_field"
  }
}
indexer {
  type: "Elasticsearch"
  routingField: "routing_field"
}

Join Field Support (Detailed)

Elasticsearch’s join field allows parent-child relationships within a single index. The ElasticsearchIndexer supports this via the elastic.join config block:

When isChild: true, the indexer adds a join field to each document with the child relation name and sets routing to the parent’s ID. Elasticsearch requires parent and child documents to be on the same shard, so indexer.routingField should also be set to the same field as parentDocumentIdSource.

Routing and Versioning

Same as OpenSearch: supports indexer.routingField for custom shard routing and indexer.versionType for optimistic concurrency control using Kafka offsets in distributed mode.

Partial Update Mode

When update: true, documents are sent as partial updates (doc-as-upsert) rather than full index operations. Same behavior as the OpenSearch Indexer.

Differences from OpenSearch Indexer

• No retry support — ElasticsearchIndexer does not wrap failures as IndexerRetryableException. The base class retry machinery will not trigger for Elasticsearch failures. If retries are needed, configure them at the Elasticsearch client or load balancer level.
• No indexOverrideField support — All documents are sent to the single configured index. You cannot route documents to different indices within the same batch.
• Child documents — The code iterates attached children but does not currently add them to the indexed document (this is a known TODO). Use emitted children (separate documents) instead of attached children if you need child documents indexed in Elasticsearch.

Deletion Support

The ElasticsearchIndexer supports three deletion mechanisms:

Delete by ID: When a document is marked with deletionMarkerField/deletionMarkerFieldValue (and does not also have deleteByFieldField/deleteByFieldValue), the Indexer issues a bulk delete-by-ID operation for that document.

Delete by query: When a document is marked for deletion AND contains both deleteByFieldField and deleteByFieldValue, the Indexer issues a _delete_by_query request. The deleteByFieldField value on the document identifies which Elasticsearch field to query, and the deleteByFieldValue value identifies the terms to match. Multiple delete-by-query documents in a batch are combined into a single bool/should query.

Batch ordering: Within a batch, if the same document ID appears in both an upload and a delete operation, the last occurrence wins. This allows a batch to contain both updates and deletions for the same logical entity.

indexer {
  type: "Elasticsearch"
  deletionMarkerField: "is_deleted"
  deletionMarkerFieldValue: "true"
  # Optional: for query-based deletions
  deleteByFieldField: "delete_target_field"
  deleteByFieldValue: "delete_target_value"
}

Troubleshooting

Join field errors: Ensure your Elasticsearch index mapping includes the join field with the correct parent and child relation names. The joinFieldName in config must match the mapping exactly.

Routing errors with joins: Parent and child documents must be on the same shard. Set indexer.routingField to the field containing the parent ID.

“index not found”: The target index must exist before indexing. Create it manually or via an index template.

6.4 - CSV Indexer

com.kmwllc.lucille.indexer.CSVIndexer

Config block: csv { ... }

indexer { type: "CSV" }
csv {
  path: "./output.csv"
  columns: ["id", "title", "body", "published_at"]
}

Limitations: CSVIndexer does not support indexer.indexOverrideField.

Column Ordering and Field Selection

The columns list determines both which fields are written and their column order. Fields not listed in columns are silently omitted from the output. The document’s id field is not automatically included — add it to columns explicitly if you want it in the CSV.

Multi-Valued Fields

When a document field is multi-valued, CSVIndexer writes the list’s toString() representation (e.g., [val1, val2]). This is a lossy representation — the square brackets and commas become part of the string. If you need a different delimiter or format for multi-valued fields, flatten them in a pipeline stage before indexing.

Append Mode

When append: true, the file is opened in append mode. Set includeHeader: false when appending to avoid duplicate header rows:

csv {
  path: "./output.csv"
  columns: ["id", "title", "body"]
  append: true
  includeHeader: false
}

Deletion Not Supported

CSVIndexer logs a warning if deletionMarkerField or deleteByFieldField are configured but does not perform any deletion. Documents marked for deletion are written as regular rows.

Directory Creation

CSVIndexer creates parent directories automatically if they don’t exist. You can specify a path like ./output/results/data.csv without creating the directories first.

Use Cases

CSVIndexer is primarily useful for:

• Testing — Verify pipeline output without a search backend.
• Debugging — Inspect what fields and values reach the indexer after pipeline processing.
• Exporting — Produce a file for import into another system.

It is not intended for production search indexing.

6.5 - NopIndexer (No-op)

com.kmwllc.lucille.indexer.NopIndexer

Discards all documents without sending them anywhere. Useful for testing pipelines when indexing output is not needed. Equivalent to setting indexer.sendEnabled: false on any other indexer.

indexer { type: "Nop" }

When to Use

• Testing pipeline logic — Run your pipeline end-to-end without needing a search backend running.
• Validating config — Confirm that connectors, stages, and the indexer block are configured correctly before pointing at a real backend.
• Benchmarking pipeline throughput — Measure how fast your pipeline processes documents in isolation, without indexer latency as a factor.

Equivalence with sendEnabled: false

NopIndexer and indexer.sendEnabled: false on another indexer are functionally identical — both discard documents after the pipeline processes them. The difference is that NopIndexer doesn’t require configuring a backend-specific block at all:

# These are equivalent:

# Option 1: NopIndexer
indexer { type: "Nop" }

# Option 2: sendEnabled on a real indexer
indexer { type: "Solr", sendEnabled: false }
solr { url: "http://localhost:8983/solr/my-collection" }

Use in RunType.TEST

Runner.runInTestMode() does not use NopIndexer. Instead, it creates the configured indexer with bypass=true, which causes sendToIndex() to skip the actual backend call while still running the full indexer loop (batching, events, metrics). This means your test config still needs a valid indexer block — but the backend does not need to be running. If you want to avoid configuring a backend block entirely in test configs, use NopIndexer explicitly:

indexer { type: "Nop" }

6.6 - Pinecone Indexer

com.kmwllc.lucille.pinecone.indexer.PineconeIndexer

Indexes vector embeddings into Pinecone. Config block: pinecone { ... }

Maven dependency:

<dependency>
  <groupId>com.kmwllc</groupId>
  <artifactId>lucille-pinecone</artifactId>
  <version>${lucille.version}</version>
</dependency>

indexer {
  class: "com.kmwllc.lucille.pinecone.indexer.PineconeIndexer"
  deletionMarkerField: "is_deleted"
  deletionMarkerFieldValue: "true"
}

pinecone {
  apiKey: ${PINECONE_API_KEY}
  index: "my-index"
  vectorField: "content_vector"
  namespace: "default"
}

Namespace Routing

When namespaces is configured, it is a map of namespace names to embedding field names. Each document is upserted into every configured namespace using the corresponding embedding field. This enables multi-vector indexing — for example, title embeddings in one namespace and body embeddings in another:

pinecone {
  apiKey: ${PINECONE_API_KEY}
  index: "my-index"
  namespaces: {
    "title-ns": "title_vector"
    "body-ns": "body_vector"
  }
}

When namespaces is not set, all documents go to the default namespace using defaultEmbeddingField.

defaultEmbeddingField vs. namespaces

Two modes are available:

• Single namespace — Set defaultEmbeddingField to the name of the vector field. All documents are upserted into the default namespace.
• Multi namespace — Set namespaces as a map of namespace → embedding field. Documents are upserted into each namespace with the corresponding vector.

At least one of these must be set when uploading documents. If both are omitted, the indexer throws at upload time.

Metadata Fields

Only fields listed in metadataFields are sent as Pinecone metadata alongside the vector. All other document fields are ignored. Metadata values are converted to strings.

pinecone {
  apiKey: ${PINECONE_API_KEY}
  index: "my-index"
  defaultEmbeddingField: "content_vector"
  metadataFields: ["title", "category", "source_url"]
}

Upsert vs. Update Mode

pinecone {
  mode: "update"
  // ...
}

Batch Size Limit

Pinecone’s API limits batches to 1000 vectors or 2MB, whichever is reached first. The indexer enforces the 1000-vector limit at startup — if indexer.batchSize exceeds 1000, the indexer throws an exception and refuses to start.

Higher-dimensional vectors may hit the 2MB limit at counts lower than 1000. If this happens, the Pinecone API returns an error at runtime. Reduce indexer.batchSize accordingly.

Deletion

Deletion is by ID only (uses deletionMarkerField / deletionMarkerFieldValue). Delete-by-query is not supported.

When namespaces is configured, deletion is issued to all configured namespaces.

Troubleshooting

“Maximum batch size for Pinecone is 1000”: Reduce indexer.batchSize to 1000 or lower.

API key invalid: Verify PINECONE_API_KEY is set and the key has write access to the target index.

Index not found: The Pinecone index must be created via the Pinecone console or API before running Lucille.

Vector dimension mismatch: The vector field on each document must have the same number of dimensions as the Pinecone index was created with. A mismatch causes a runtime error from the Pinecone API.

6.7 - Weaviate Indexer

com.kmwllc.lucille.weaviate.indexer.WeaviateIndexer

Indexes documents into Weaviate. Config block: weaviate { ... }

Maven dependency:

<dependency>
  <groupId>com.kmwllc</groupId>
  <artifactId>lucille-weaviate</artifactId>
  <version>${lucille.version}</version>
</dependency>

indexer {
  class: "com.kmwllc.lucille.weaviate.indexer.WeaviateIndexer"
}

weaviate {
  apiKey: ${WEAVIATE_API_KEY}
  host: "my-cluster.weaviate.network"
  className: "Article"
  idDestinationName: "lucille_id"
  vectorField: "content_vector"
}

Note on IDs: Weaviate requires UUIDs for its internal id field. The WeaviateIndexer generates a UUID from the document’s Lucille ID and stores the original ID under idDestinationName so it can be retrieved later.

UUID Generation

Weaviate requires UUIDs for its internal id field. The WeaviateIndexer generates a deterministic UUID from the document’s Lucille ID using UUID.nameUUIDFromBytes(id.getBytes()). This means:

• The same Lucille ID always maps to the same Weaviate UUID, enabling idempotent upserts.
• The original Lucille ID is stored under idDestinationName (default: "id_original") so it can be retrieved later.

Deletion Not Supported

WeaviateIndexer logs a warning if deletionMarkerField or deleteByFieldField are configured but does not perform deletions. Documents marked for deletion are indexed as regular objects. This is a known limitation.

Vector Field Handling

If vectorField is set and the document has that field, the vector is sent alongside the object properties. The vector field is removed from the properties map to avoid storing it twice (once as a vector, once as a property).

If vectorField is omitted, no vector is sent. This is useful when Weaviate is configured to generate its own embeddings via vectorizer modules (e.g., text2vec-openai).

Consistency Level

All writes use ConsistencyLevel.ALL (strongest consistency). This is not currently configurable. In a multi-node Weaviate cluster, this means all replicas must acknowledge the write before it is considered successful.

Connection Timeouts

The Weaviate client is configured with 6-second timeouts for connection, read, and write operations. These are not currently configurable via the Lucille config. If you experience timeout errors with large batches or slow networks, reduce indexer.batchSize.

Class/Schema Requirements

The className parameter determines which Weaviate class (schema type) objects are created under. The class must already exist in the Weaviate schema — the indexer does not create it. If the class doesn’t exist, the batch write will fail.

Troubleshooting

AuthException (“Couldn’t connect to Weaviate instance”): The API key is invalid or the host is unreachable. Verify WEAVIATE_API_KEY and that the host value is correct (hostname only, no https:// prefix).

Class not found in schema: Create the Weaviate class via the Weaviate console or REST API before running Lucille.

Vector dimension mismatch: If using a Weaviate vectorizer module, ensure the vector field dimensions match what the module expects. If sending vectors directly, ensure they match the class’s configured vector dimensions.

Timeout errors: Reduce indexer.batchSize to send smaller batches. The 6-second timeout is fixed.

7 - Cookbooks

Lucille cookbooks are practical, step-by-step guides for common ingestion patterns.

Available Cookbooks

• File Ingestion — Ingest files from the local filesystem, Amazon S3, Azure Blob Storage, and Google Cloud Storage. Covers CSV, JSON, XML, incremental mode, tombstone deletions, and Tika text extraction.

• Vector Search — Build end-to-end vector search pipelines: chunk text, generate embeddings (OpenAI or local via JLama), and index into Pinecone or Weaviate.

• RSS Ingestion — Ingest RSS feeds into CSV or OpenSearch, including incremental mode.

7.1 - File Ingestion Cookbook

This cookbook covers common file ingestion patterns using Lucille’s FileConnector.

Recipe 1: Ingest CSV Files from Local Filesystem

Read all .csv files in a directory and index each row as a document into OpenSearch.

connectors: [
  {
    name: "csv-connector"
    class: "com.kmwllc.lucille.connector.FileConnector"
    pipeline: "csv-pipeline"
    paths: ["/data/csvfiles"]

    fileHandlers: {
      csv {
        filenameField: "source_file"
        docIdPrefix: "row-"
      }
    }

    filterOptions: {
      includes: [".*\\.csv$"]
    }
  }
]

pipelines: [
  {
    name: "csv-pipeline"
    stages: [
      {
        class: "com.kmwllc.lucille.stage.TrimWhitespace"
        fields: ["title", "description"]
      }
    ]
  }
]

indexer { type: "opensearch" }

opensearch {
  url: "https://localhost:9200"
  index: "my-docs"
  acceptInvalidCert: true
}

Recipe 2: Ingest JSON Files

Read .json or .jsonl files. Each top-level JSON object (or each line in a .jsonl file) becomes a Document.

connectors: [
  {
    name: "json-connector"
    class: "com.kmwllc.lucille.connector.FileConnector"
    pipeline: "json-pipeline"
    paths: ["/data/jsonfiles"]

    fileHandlers: {
      json {}
    }
  }
]

For .jsonl files (one JSON object per line), no extra configuration is needed — the JSONFileHandler handles both formats automatically.

Recipe 3: Ingest XML Files

Extract records from XML files using an XPath expression.

connectors: [
  {
    name: "xml-connector"
    class: "com.kmwllc.lucille.connector.FileConnector"
    pipeline: "xml-pipeline"
    paths: ["/data/xmlfiles"]

    fileHandlers: {
      xml {
        chunkPath: "//record"
      }
    }
  }
]

Each XML element matching //record becomes a separate Lucille Document.

Recipe 4: Ingest from Amazon S3

connectors: [
  {
    name: "s3-connector"
    class: "com.kmwllc.lucille.connector.FileConnector"
    pipeline: "my-pipeline"
    paths: ["s3://my-bucket/data/"]

    s3 {
      accessKeyId: ${?AWS_ACCESS_KEY_ID}
      secretAccessKey: ${?AWS_SECRET_ACCESS_KEY}
      region: ${?AWS_DEFAULT_REGION}
    }

    fileHandlers: {
      csv {}
    }
  }
]

For paths with special characters (e.g., spaces), percent-encode the URI: s3://my-bucket/folder%20with%20spaces/.

Recipe 5: Ingest from Azure Blob Storage

connectors: [
  {
    name: "azure-connector"
    class: "com.kmwllc.lucille.connector.FileConnector"
    pipeline: "my-pipeline"
    paths: ["https://mystorageaccount.blob.core.windows.net/my-container/"]

    azure {
      connectionString: ${?AZURE_CONNECTION_STRING}
    }
  }
]

Alternatively, authenticate with account name and key:

azure {
  accountName: ${?AZURE_ACCOUNT_NAME}
  accountKey: ${?AZURE_ACCOUNT_KEY}
}

Recipe 6: Ingest from Google Cloud Storage

connectors: [
  {
    name: "gcs-connector"
    class: "com.kmwllc.lucille.connector.FileConnector"
    pipeline: "my-pipeline"
    paths: ["gs://my-bucket/data/"]

    gcp {
      pathToServiceKey: ${?GCP_SERVICE_KEY_PATH}
    }
  }
]

Recipe 7: Incremental Ingest (Only New or Modified Files)

Use state tracking to skip files that have not changed since the last run.

connectors: [
  {
    name: "incremental-connector"
    class: "com.kmwllc.lucille.connector.FileConnector"
    pipeline: "my-pipeline"
    paths: ["/data/files"]
    publishMode: "INCREMENTAL"

    # JDBC state database (H2 embedded is the default; Derby also works)
    state {
      driver: "org.h2.Driver"
      connectionString: "jdbc:h2:./lucille-state"
    }

    filterOptions: {
      # Only process files that were last published more than 1 hour ago
      lastPublishedCutoff: "1h"
    }
  }
]

Notes on incremental mode:

• FULL mode (the default) republishes every file on every run.
• INCREMENTAL mode requires a state database and only processes files that are new or modified since the last run.
• The state database tracks when each file was last published by Lucille.

Recipe 8: Tombstone Deletions

Automatically detect deleted files and issue deletes against the search backend.

connectors: [
  {
    name: "connector-with-tombstones"
    class: "com.kmwllc.lucille.connector.FileConnector"
    pipeline: "my-pipeline"
    paths: ["/data/files"]
    publishMode: "INCREMENTAL"
    sendTombstones: true

    state {
      driver: "org.h2.Driver"
      connectionString: "jdbc:h2:./lucille-state"
    }
  }
]

indexer {
  type: "opensearch"
  deletionMarkerField: "file_expired"
  deletionMarkerFieldValue: "true"
}

When a file is deleted from the filesystem, Lucille creates a tombstone Document with file_expired: true. The OpenSearchIndexer sees this marker and issues a delete request against the search backend.

Recipe 9: Extract Text from PDF, Office, and Other Formats (Tika)

Use the lucille-tika plugin to extract text from arbitrary file formats. Set getFileContent: true in fileOptions to read raw bytes into file_content, then pass it to the TextExtractor stage.

connectors: [
  {
    name: "docs-connector"
    class: "com.kmwllc.lucille.connector.FileConnector"
    pipeline: "docs-pipeline"
    paths: ["/data/documents"]

    fileOptions: {
      getFileContent: true
    }
  }
]

pipelines: [
  {
    name: "docs-pipeline"
    stages: [
      {
        class: "com.kmwllc.lucille.tika.stage.TextExtractor"
        source: "file_content"
        dest: "extracted_text"
      }
    ]
  }
]

Requires the lucille-tika Maven dependency. See TextExtractor in All Stages for setup.

Filter Options Reference

Use filterOptions to control which files are processed:

filterOptions: {
  # Only process files whose names match these patterns (regex)
  includes: [".*\\.pdf$", ".*\\.docx$"]

  # Skip files whose names match these patterns (regex)
  excludes: [".*\\.DS_Store$", ".*\\.tmp$"]

  # Only process files modified within the last 3 days
  lastModifiedCutoff: "3d"

  # Only process files not published by Lucille in the last 6 hours (requires state)
  lastPublishedCutoff: "6h"
}

Duration values accept: s (seconds), m (minutes), h (hours), d (days).

File Options Reference

fileOptions controls traversal behavior. fileHandlers (a separate block) configures which file types to process and how:

fileOptions: {
  # Read file bytes into the file_content field (slow; downloads cloud files)
  getFileContent: true

  # Process zip/tar archives by extracting their contents
  handleArchivedFiles: true

  # Process gzip-compressed files
  handleCompressedFiles: true

  # Move files to this path after successful processing (local files only)
  moveToAfterProcessing: "/data/processed"

  # Move files to this path if an error occurs (local files only)
  moveToErrorFolder: "/data/errors"
}

fileHandlers: {
  csv { ... }
  json { ... }
  xml { ... }
}

7.2 - Vector Search Cookbook

This cookbook shows how to build end-to-end vector search pipelines with Lucille.

Overview

A typical vector search pipeline:

1. Read source documents (files, database, RSS, etc.)
2. Extract text from each document if needed (Tika for PDF/Office files)
3. Chunk long text into smaller pieces suitable for embedding
4. Embed each chunk using an embedding model
5. Index the vectors into a vector database (Pinecone or Weaviate)

Recipe 1: CSV → OpenAI Embeddings → Pinecone

Index a CSV of articles into Pinecone using the OpenAI Embeddings API.

Prerequisites:

• lucille-pinecone plugin on the classpath
• An OpenAI API key in the OPENAI_API_KEY environment variable
• A Pinecone index configured to match the embedding model’s output dimensions

connectors: [
  {
    name: "article-connector"
    class: "com.kmwllc.lucille.connector.FileConnector"
    pipeline: "embed-pipeline"
    paths: ["/data/articles.csv"]

    fileHandlers: {
      csv {
        docIdPrefix: "article-"
      }
    }
  }
]

pipelines: [
  {
    name: "embed-pipeline"
    stages: [
      # Combine title and body into a single text field for embedding
      {
        class: "com.kmwllc.lucille.stage.Concatenate"
        source: ["title", "body"]
        dest: "text_to_embed"
        delimiter: " "
      },
      # Generate a vector embedding using OpenAI
      {
        class: "com.kmwllc.lucille.stage.OpenAIEmbed"
        source: "text_to_embed"
        dest: "content_vector"
        modelName: "text-embedding-3-small"
        apiKey: ${OPENAI_API_KEY}
        embedDocument: true
        embedChildren: false
      }
    ]
  }
]

indexer {
  class: "com.kmwllc.lucille.pinecone.indexer.PineconeIndexer"
}

pinecone {
  apiKey: ${PINECONE_API_KEY}
  index: "articles-index"
  vectorField: "content_vector"
  namespace: "articles"
}

Recipe 2: PDF Files → Tika → Chunk → Local Embeddings (JLama) → Pinecone

Index PDF documents without sending data to an external API. The JlamaEmbed stage generates embeddings locally.

Prerequisites:

• lucille-tika plugin (for text extraction)
• lucille-jlama plugin (for local embedding)
• lucille-pinecone plugin (for indexing)

connectors: [
  {
    name: "pdf-connector"
    class: "com.kmwllc.lucille.connector.FileConnector"
    pipeline: "pdf-embed-pipeline"
    paths: ["/data/pdfs"]

    filterOptions: {
      includes: [".*\\.pdf$"]
    }

    fileOptions: {
      getFileContent: true
    }
  }
]

pipelines: [
  {
    name: "pdf-embed-pipeline"
    stages: [
      # Extract text from PDF bytes using Apache Tika
      {
        class: "com.kmwllc.lucille.tika.stage.TextExtractor"
        source: "file_content"
        dest: "extracted_text"
      },
      # Split long text into smaller chunks (each chunk → child document)
      {
        class: "com.kmwllc.lucille.stage.ChunkText"
        source: "extracted_text"
        dest: "text"
        chunkingMethod: "sentence"
        chunksToMerge: 5
        chunksToOverlap: 1
        cleanChunks: true
        characterLimit: 2000
      },
      # Generate embeddings locally (no external API call) — only embed child chunks
      {
        class: "com.kmwllc.lucille.jlama.stage.JlamaEmbed"
        source: "text"
        dest: "content_vector"
        modelPath: "/models/all-MiniLM-L6-v2"
        embedDocument: false
        embedChildren: true
      }
    ]
  }
]

indexer {
  class: "com.kmwllc.lucille.pinecone.indexer.PineconeIndexer"
}

pinecone {
  apiKey: ${PINECONE_API_KEY}
  index: "pdf-index"
  vectorField: "content_vector"
}

Recipe 3: Parquet → Embeddings → Pinecone

Read pre-computed vector embeddings from a Parquet file (e.g., embeddings generated by an external model) and index them directly.

Prerequisites:

• lucille-parquet plugin
• lucille-pinecone plugin

connectors: [
  {
    name: "parquet-connector"
    class: "com.kmwllc.lucille.parquet.connector.ParquetConnector"
    pipeline: "vector-pipeline"
    path: "/data/embeddings.parquet"
    idField: "doc_id"
  }
]

pipelines: [
  {
    name: "vector-pipeline"
    stages: []
  }
]

indexer {
  class: "com.kmwllc.lucille.pinecone.indexer.PineconeIndexer"
}

pinecone {
  apiKey: ${PINECONE_API_KEY}
  index: "my-vectors"
  vectorField: "embedding"
}

Recipe 4: Chunking Strategy

The ChunkText Stage splits a long text field into chunks and emits each chunk as a child document. The child documents flow through all downstream stages and are indexed independently — the parent document is also indexed (unless dropped).

{
  class: "com.kmwllc.lucille.stage.ChunkText"
  source: "body"
  dest: "text"              # field name in child docs
  chunkingMethod: "sentence"
  chunksToMerge: 5          # merge 5 sentences into each final chunk
  chunksToOverlap: 1        # 1 sentence of overlap between adjacent chunks
  cleanChunks: true         # strip newlines and trim whitespace
  characterLimit: 2000      # hard cap per chunk
}

After chunking, each child document has:

• id: {parent-id}-chunk-{n} (e.g., doc-42-chunk-0, doc-42-chunk-1, …)
• parent_id: ID of the parent document
• chunk_number: zero-based index
• total_chunks: total number of chunks from this parent
• offset: character offset of this chunk in the original text
• length: character length of this chunk
• text (or whatever dest is set to): the chunk text

Parent document fields are not automatically copied to child documents.

Recipe 5: Conditional Embedding

Only generate embeddings for documents that have content, skipping short or empty documents:

{
  class: "com.kmwllc.lucille.stage.OpenAIEmbed"
  source: "content"
  dest: "content_vector"
  modelName: "text-embedding-3-small"
  apiKey: ${OPENAI_API_KEY}
  embedDocument: true
  embedChildren: false
  conditions: [
    { fields: ["content"], operator: "must" }
  ]
}

Recipe 6: Database → Ollama Summarization → Embeddings → OpenSearch

Enrich database records with LLM-generated summaries before embedding.

connectors: [
  {
    name: "db-connector"
    class: "com.kmwllc.lucille.connector.jdbc.DatabaseConnector"
    pipeline: "enrich-embed"
    driver: "org.postgresql.Driver"
    connectionString: "jdbc:postgresql://localhost:5432/mydb"
    sql: "SELECT id, title, body FROM documents"
    idField: "id"
  }
]

pipelines: [
  {
    name: "enrich-embed"
    stages: [
      # Generate a concise summary using a local Ollama model
      {
        class: "com.kmwllc.lucille.stage.PromptOllama"
        hostURL: "http://localhost:11434"
        modelName: "llama3"
        systemPrompt: "Summarize the following document in 2-3 sentences. Output only a JSON object with a single field: summary."
        fields: ["body"]
      },
      # Embed the summary
      {
        class: "com.kmwllc.lucille.stage.OpenAIEmbed"
        source: "summary"
        dest: "summary_vector"
        modelName: "text-embedding-3-small"
        apiKey: ${OPENAI_API_KEY}
        embedDocument: true
        embedChildren: false
      }
    ]
  }
]

indexer { type: "opensearch" }

opensearch {
  url: "https://localhost:9200"
  index: "enriched-docs"
  acceptInvalidCert: true
}

Model Comparison

Tips

• Batch size: Embedding API calls benefit from larger batches. The default indexer.batchSize (100) is a good starting point.
• Rate limiting: If you hit OpenAI rate limits, reduce worker.threads or add retry logic.
• Model dimensions: Your Pinecone/Weaviate index dimensions must match the embedding model’s output (e.g., text-embedding-3-small outputs 1536 dimensions).
• Testing: Use RandomVector Stage to generate fake embeddings during development without API calls.

7.3 - RSS Cookbook

RSS to CSV

Let’s say we wanted to read from an RSS feed into a CSV using Lucille. We can set this up in the following manner:

1. Create a .conf file to configure Lucille
2. Specify the RSS connector in the connectors section of the config:

connectors: [
  {
    name: "RSSConnector"
    pipeline: "rssPipeline"
    class: "com.kmwllc.lucille.connector.RSSConnector"
    rssURL: "https://www.cnbc.com/id/15837362/device/rss/rss.html"
  }
]

There are a few additional configuration options that we won’t use here, but are useful:

    useGuidForDocID: true       # defaults to true; set false to use UUID as ID instead
    pubDateCutoff: "24h"        # only publish items from the last 24 hours
    runDuration: "1h"           # run incrementally for 1 hour total
    refreshIncrement: "5m"      # re-fetch the feed every 5 minutes

Your pipeline name can be whatever you want. For our URL, we chose CNBC’s RSS feed.

3. Now we define what stages we would like to use to process our documents from the feed. To give context as to what these stages are doing:

News items in an RSS feed often have some article metadata, and then a link to the actual meat of the article in HTML as a field.

• The fetchURI stage allows us to grab the actual content of our associated news article.
• The ApplyJSoup stage parses that content into fields that will exist in addition to our article metadata from the RSS feed. These fields include the body, bullet points, and the header.

pipelines: [
  {
    name: "rssPipeline"
    stages: [
      {
        name: "fetchURI",
        class: "com.kmwllc.lucille.stage.FetchUri"
        source: "link"
        dest: "content"
      }
      {
        name: "ApplyJSoup"
        class: "com.kmwllc.lucille.stage.ApplyJSoup"
        byteArrayField: "content"
        destinationFields: {
          paragraphTexts: {
            type: "text",
            selector: ".ArticleBody-articleBody p"
          }
          bulletPoints: {
            type: "text",
            selector: ".RenderKeyPoints-list li"
          }
          headline: {
            type: "text"
            selector: "h1"
          }
        }
      }
    ]
  }
]

4. We can index these documents into whatever we’d like. Here, we might decide to just print them to a CSV:

indexer: {
  type: "csv"
}

csv: {
  path: "./rss_results.csv"
  columns: ["id", "link", "title", "description", "paragraphTexts",
    "bulletPoints", "headline"]
}

Here is the full config file:

connectors: [
  {
    name: "RSSConnector"
    pipeline: "rssPipeline"
    class: "com.kmwllc.lucille.connector.RSSConnector"
    rssURL: "https://www.cnbc.com/id/15837362/device/rss/rss.html"
  }
]

pipelines: [
  {
    name: "rssPipeline"
    stages: [
      {
        name: "fetchURI",
        class: "com.kmwllc.lucille.stage.FetchUri"
        source: "link"
        dest: "content"
      }
      {
        name: "ApplyJSoup"
        class: "com.kmwllc.lucille.stage.ApplyJSoup"
        byteArrayField: "content"
        destinationFields: {
          paragraphTexts: {
            type: "text",
            selector: ".ArticleBody-articleBody p"
          }
          bulletPoints: {
            type: "text",
            selector: ".RenderKeyPoints-list li"
          }
          headline: {
            type: "text"
            selector: "h1"
          }
        }
      }
    ]
  }
]

indexer: {
  type: "csv"
}

csv: {
  path: "./rss_results.csv"
  columns: ["id", "link", "title", "description", "paragraphTexts",
    "bulletPoints", "headline"]
}

The CSV file will thus be saved on disk.

RSS to OpenSearch

We might also choose to index into another destination, like an OpenSearch index. Here’s an example. Replace the config with this:

connectors: [
  {
    name: "RSSConnector"
    pipeline: "rssPipeline"
    class: "com.kmwllc.lucille.connector.RSSConnector"
    rssURL: "https://www.cnbc.com/id/15837362/device/rss/rss.html"
    refreshIncrement: "60s"
    runDuration: "1h"
  }
]

pipelines: [
  {
    name: "rssPipeline"
    stages: []
  }
]

indexer: {
  type: "opensearch"
}

opensearch: {
  url: <Your OpenSearch URL>
  index: "rss-index"
  acceptInvalidCert: true
}

You’ll notice we’re using incremental mode now, with a refreshIncrement of 60s and a runDuration of 1h. This means that every item in the feed will be indexed on the initial run, and then Lucille will continue to take in new items that pop up every 60 seconds for 1 hour total.

Run Lucille again. Here’s what 3 of our documents look like after being indexed into OpenSearch:

GET /rss-index/_search
{
  "size": 3
}

{
  "took": 15,
  "timed_out": false,
  "_shards": {
    "total": 1,
    "successful": 1,
    "skipped": 0,
    "failed": 0
  },
  "hits": {
    "total": {
      "value": 30,
      "relation": "eq"
    },
    "max_score": 1,
    "hits": [
      {
        "_index": "rss-index",
        "_id": "108275704",
        "_score": 1,
        "_source": {
          "id": "108275704",
          "guid": "108275704",
          "isPermaLink": false,
          "link": "https://www.cnbc.com/2026/03/09/watch-live-trump-press-conference-iran-war-oil-hormuz-doral.html",
          "title": "Watch live: Trump holds press conference as Iran war fallout roils oil market",
          "pubDate": "2026-03-09T21:13:25Z",
          "run_id": "cb234bd2-fdf7-4b88-be13-20de36cd059e"
        }
      },
      {
        "_index": "rss-index",
        "_id": "108275619",
        "_score": 1,
        "_source": {
          "id": "108275619",
          "description": "The OpenAI deal fallout exposes the fundamental danger of being the most leveraged player.",
          "guid": "108275619",
          "isPermaLink": false,
          "link": "https://www.cnbc.com/2026/03/09/oracle-is-building-yesterdays-data-centers-with-tomorrows-debt.html",
          "title": "Oracle is building yesterday's data centers with tomorrow's debt",
          "pubDate": "2026-03-09T20:52:19Z",
          "run_id": "cb234bd2-fdf7-4b88-be13-20de36cd059e"
        }
      },
      {
        "_index": "rss-index",
        "_id": "108275649",
        "_score": 1,
        "_source": {
          "id": "108275649",
          "description": "U.S. stock market indexes rose on the heels of reported comments by President Donald Trump.",
          "guid": "108275649",
          "isPermaLink": false,
          "link": "https://www.cnbc.com/2026/03/09/trump-iran-war-end.html",
          "title": "Trump says Iran 'war is very complete,' talks to Putin, reports say",
          "pubDate": "2026-03-09T21:32:43Z",
          "run_id": "cb234bd2-fdf7-4b88-be13-20de36cd059e"
        }
      }
    ]
  }
}

Using other indexers with the RSS connector follows much the same pattern.