ACID Lakehouse Streaming, Parquet Storage & Dynamic Schema Evolution: Announcing the Apache Iceberg Output Connector

1. Bridging Enterprise Repositories Directly to the Lakehouse

In modern enterprise data platforms, data architectures often face a divide: enterprise content management (ECM) repositories, document management systems, and workflow engines live in isolated operational silos, while analytical and machine learning workloads thrive on open data lakehouse table formats like Apache Iceberg.

Traditional search and vector databases (such as Elasticsearch, OpenSearch, Solr, pgvector, or Milvus) are unmatched for low-latency top-k semantic search and hybrid kNN retrieval. However, enterprises increasingly require long-term compliance archiving, petabyte-scale historical reporting, multi-engine federated SQL querying (via DuckDB, Trino, and Apache Spark), and immutable snapshot time travel.

With the release of oc-iceberg-output-connector, OpenCrawling becomes the first enterprise ingestion framework capable of streaming unstructured documents, rich metadata attributes, extracted text, and normalized Open Ingestion Standard (OIS) Zero-Trust Access Control Lists (ACLs) directly into Apache Iceberg Parquet tables with transactional snapshot isolation.

2. Direct Batching vs. Decoupled Kafka Streaming

The Iceberg Output Connector provides architectural flexibility across both standalone and horizontally scaled deployments:

• Direct Synchronous Path (IcebergOutputConnector): Implements OpenCrawling's core OutputConnector SPI. Suitable for standalone batch jobs or lightweight containers where the crawler writes directly to the Iceberg catalog and Parquet storage using commitBatch(...).
• Decoupled Asynchronous Path (IcebergStoreWriterConsumer): Designed for production high-throughput clusters. The crawler offloads large binaries to the Claim Check Store (MinIO, AWS S3, or Apache Ozone) and publishes an IngestionMessage to Apache Kafka. The decoupled writer consumes from the topic with configurable worker concurrency (default 3), resolves content streams, applies dynamic schema evolution, appends Parquet records, and commits ACID table snapshots.

3. Pluggable Enterprise Catalogs via IcebergCatalogFactory

Rather than hardcoding a specific catalog backend, IcebergCatalogFactory instantiates the appropriate Iceberg Catalog based on the spring.opencrawling.output.iceberg.catalog-type property:

• rest (Recommended): Connects to modern REST Catalog servers (e.g., Apache Iceberg REST Fixture, Tabular, Apache Polaris, Lakekeeper).
• hive: Connects to existing enterprise Hive Metastores via Thrift (iceberg-hive-metastore).
• hadoop: Direct directory-based catalog over HDFS, cloud buckets, or local filesystems.
• glue: Native AWS Glue Catalog integration leveraging the AWS SDK v2 client.
• jdbc: Relational catalog integration using PostgreSQL, MySQL, or SQLite backends.
• in-memory: Ephemeral catalog for fast unit tests and zero-dependency verification.

4. Canonical Lakehouse Schema & Zero-Trust OIS ACL Enforcement

On startup, IcebergTableManager provisions the target Iceberg namespace and table with the strongly typed CANONICAL_SCHEMA:

// Canonical OpenCrawling Lakehouse Schema for Apache Iceberg
Types.NestedField.required(1,  "id",                   Types.StringType.get(), "Unique record UUID"),
Types.NestedField.required(2,  "document_id",          Types.StringType.get(), "Canonical document ID"),
Types.NestedField.optional(3,  "source_type",          Types.StringType.get(), "Repository connector type"),
Types.NestedField.optional(4,  "document_uri",         Types.StringType.get(), "Original repository URI"),
Types.NestedField.optional(5,  "title",                Types.StringType.get(), "Document title or filename"),
Types.NestedField.optional(6,  "mime_type",            Types.StringType.get(), "MIME content type"),
Types.NestedField.optional(7,  "content_length",       Types.LongType.get(),   "Binary length in bytes"),
Types.NestedField.optional(8,  "sha256_checksum",      Types.StringType.get(), "SHA-256 binary checksum"),
Types.NestedField.optional(9,  "claim_check_uri",      Types.StringType.get(), "Claim-Check object storage URI"),
Types.NestedField.optional(10, "extracted_text",       Types.StringType.get(), "Extracted plain text"),
Types.NestedField.optional(11, "extracted_entities",   Types.MapType.ofOptional(12, 13, Types.StringType.get(), Types.StringType.get())),
Types.NestedField.optional(14, "custom_metadata",      Types.MapType.ofOptional(15, 16, Types.StringType.get(), Types.StringType.get())),
Types.NestedField.optional(17, "security_owner",       Types.StringType.get(), "Primary security owner"),
Types.NestedField.optional(18, "security_inheritance", Types.BooleanType.get(),"Inheritance flag"),
Types.NestedField.optional(19, "allowed_principals",   Types.ListType.ofOptional(20, Types.StringType.get()), "Zero-Trust ACL Allow"),
Types.NestedField.optional(21, "denied_principals",    Types.ListType.ofOptional(22, Types.StringType.get()), "Zero-Trust ACL Deny"),
Types.NestedField.required(23, "action",               Types.StringType.get(), "OIS Action: UPSERT or DELETE"),
Types.NestedField.required(24, "crawled_at",           Types.TimestampType.withoutZone(), "Crawl timestamp UTC")

5. Dynamic Runtime Schema Evolution

Enterprise documents often carry bespoke business descriptors that cannot be anticipated at design time. When contracts, invoices, or claims are crawled, Apache Tika and Spring AI extract custom entities (such as entity.vendor, entity.amount, or entity.contract_id).

When spring.opencrawling.output.iceberg.auto-schema-evolution=true is enabled, IcebergTableManager introspects incoming entity attributes. If a column is missing from the active table schema, it commits an Iceberg schema update via:

// Dynamic runtime schema evolution without table locks or data rewrites
UpdateSchema updateSchema = table.updateSchema();
for (String entityKey : entities.keySet()) {
    String colName = IcebergTableManager.sanitizeIdentifier(entityKey);
    if (currentSchema.findField(colName) == null) {
        updateSchema.addColumn(colName, Types.StringType.get(), "Dynamically extracted entity: " + entityKey);
    }
}
updateSchema.commit();
table.refresh();

Because Apache Iceberg tracks schema versions in metadata rather than rewriting underlying data files, this evolution occurs instantly with zero ingestion downtime.

6. Partition Transforms & Columnar Storage

To prevent the "small files problem" and optimize query pruning, the connector supports configurable partition strategies:

• YEAR_MONTH (Default): Employs Iceberg's native partition transform on the crawled_at timestamp (builder.month("crawled_at")). This keeps data partitioned into manageable monthly directories while allowing queries with date ranges to skip irrelevant partitions completely.
• SOURCE_AND_DATE: Partitions by repository identity (source_type) and monthly crawl date.
• NONE: Leaves the table unpartitioned for smaller corporate archives.

Data files are formatted as pure Apache Parquet compressed with Zstandard (zstd) or Snappy, delivering up to 70% storage savings compared to uncompressed raw JSON while enabling vectorized reads across analytical engines.

7. Configuration Reference

All properties bind cleanly under spring.opencrawling.output.iceberg.*:

8. Testing, Decoupled Stack & DuckDB OLAP Queries

The Apache Iceberg Output Connector includes complete turnkey automated test suites:

# 1. Test standalone REST catalog interaction & dynamic schema evolution
./scripts/test-iceberg-output-connector.sh

# 2. Test full multi-service decoupled ingestion pipeline (Kafka, MinIO, REST Catalog)
./scripts/test-iceberg-output-decoupled.sh

Once ingested, you can query your lakehouse tables immediately with in-process DuckDB without running heavy analytical clusters:

-- Querying ingested enterprise documents in DuckDB
INSTALL iceberg;
LOAD iceberg;

SELECT document_id, title, action, security_owner, allowed_principals, entity_vendor
FROM iceberg_scan('s3://warehouse/enterprise_data/documents',
                  endpoint='http://localhost:9000',
                  access_key_id='admin',
                  secret_access_key='password')
WHERE 'role:lakehouse-analytics' = ANY(allowed_principals);