Skip to content

Conventions

ftm-lakehouse is convention-driven: the path layout, artifact names and freshness tags are stable contracts – third-party tools can populate or consume a lakehouse by following them, without going through this library.

Storage Layout

On-disk (or cloud storage) layout of a lakehouse storage root:

lakehouse/
└── {dataset}/
    ├── config.yml                # Dataset configuration (shards, compression, ...)
    ├── index.json                # Published dataset index with statistics
    ├── .LOCK                     # Exclusive maintenance fence
    ├── .LOCK-APPENDS/            # In-flight append markers (shared fence)
    ├── archive/                  # Content-addressed file storage
    │   └── {ch[0:2]}/{ch[2:4]}/{ch[4:6]}/{checksum}/
    │       ├── blob              # Raw file content (stored once)
    │       ├── {file_id}.json    # File metadata (one per source path)
    │       └── {origin}.txt      # Extracted text (one per engine)
    ├── statements/               # Delta Lake parquet store
    │   ├── _delta_log/
    │   └── shard={hex}/bucket={bucket}/origin={origin}/*.parquet
    ├── entities.ftm.json[.gz|.zst]   # Aggregated entities export
    ├── exports/
    │   ├── statements.csv[.gz|.zst]  # Sorted statements export
    │   ├── statistics.json           # Entity counts, facets
    │   ├── documents.csv[.gz|.zst]   # Document metadata
    │   └── documents.{origin}.csv[..] # Document metadata, one origin only
    ├── diffs/                    # Timestamped delta diff exports
    ├── versions/                 # Versioned snapshots (config, index, ...)
    │   └── YYYY/MM/{timestamp}/
    ├── tags/{tenant}/            # Freshness tags (workflow state)
    └── jobs/
        └── runs/{job_type}/{timestamp}.json

Freshness Tags

Operations use tags to track freshness and skip unnecessary work – is_latest(key, dependencies) returns True when key is newer than all its dependencies:

Tag Set by Meaning
journal/last_updated Statement writes Journal has uncommitted data
journal/last_flushed Flush operation Journal was flushed
statements/last_updated Flush / append Rows landed in the parquet store – not canonical yet
statements/last_optimized merge, on completion Canonical content changed. The clock every export, statistic and diff goes stale against
archive/last_updated File archive New file was archived
exports/statements.csv, entities.ftm.json, exports/documents.csv, exports/documents.{origin}.csv, exports/statistics.json, index.json Export operations Export target keys double as their freshness tags. The all sweep stamps every artifact it writes, so a later single-kind export sees itself up to date
operations/export/last_run export all The fused sweep ran
operations/crawl/last_run Crawl operation Last crawl execution

Compression suffixes

When a dataset configures compression (gz / zst in config.yml), the streaming export artifacts carry the codec suffix – entities.ftm.json.zst, exports/statements.csv.zst, exports/documents.csv.zst – and index.json advertises the resulting names and urls. index.json and statistics.json themselves are always plain JSON.

Diff directories stay codec-free (diffs/exports/documents.csv/), because they double as the freshness tag and diff-state key; only the files inside them carry the suffix ({timestamp}.diff.csv.zst).

Path conventions

All path construction goes through ftm_lakehouse.core.conventions.path – rendered here so the constants stay in sync with the code:

Path conventions for the FollowTheMoney data lakehouse.

The fundamental idea is to have a convention-based file system layout with well-known paths for metadata and information interchange between processing stages.

All paths are dataset-relative unless otherwise noted.

Dataset Layout

::

lakehouse/
    index.json                          # catalog index
    config.yml                          # catalog configuration
    versions/                           # versioned snapshots
        YYYY/MM/YYYY-MM-DDTHH:MM:SS/
            index.json
            config.yml

    [dataset]/
        index.json                      # dataset index
        config.yml                      # dataset configuration

        versions/                       # versioned snapshots
            YYYY/MM/...

        .LOCK                           # dataset-wide maintenance lock
        .LOCK-APPENDS/                  # in-flight append markers
        .locks/{tenant}/                 # operation-specific locks
        tags/{tenant}/                  # workflow state / cache

        archive/                        # content-addressed file storage
            ab/cd/ef/{checksum}/        # SHA256 split into segments
                blob                    # file blob (stored once)
                {file_id}.json          # metadata (one per source path)
                {origin}.txt            # extracted text (one per engine)

        statements/                     # statement store (shard-partitioned)
            shard={shard}/
                bucket={bucket}/
                    origin={origin}/
                        *.parquet

        entities.ftm.json[.zst|gzip]    # aggregated entities export

        exports/
            statistics.json             # entity counts, facets
            statements.csv[.zst|gzip]   # sorted statements
            documents.csv[.zst|gzip]    # document metadata
            documents.{origin}.csv[...] # document metadata (origin-scoped)
            graph.cypher                # neo4j export (optional)

        diffs/                          # dirs are codec-free (they double
            entities.ftm.json/          #   as freshness tags); the files
                {ts}.delta.json[.zst|gzip]         # entities delta
            exports/
                documents.csv/
                    {ts}.diff.csv[.zst|gzip]       # documents delta
                documents.{origin}.csv/
                    {ts}.diff.csv[.zst|gzip]       # origin-scoped delta

        jobs/
            runs/
                {job_type}/
                    {timestamp}.json    # job run results

ARCHIVE = 'archive' module-attribute

Base path for archive

ARCHIVE_BLOB = 'blob' module-attribute

blob filename within checksum directory

CONFIG = StoreKey('config.yml') module-attribute

user editable config filename

DIFFS = StoreKey('diffs') module-attribute

Base path for diff exports

DIFFS_DOCUMENTS = DateTimeKey(DIFFS / EXPORTS_DOCUMENTS, TS_FORMAT, EXT_DOCUMENTS_DELTA) module-attribute

Documents diff series: DIFFS_DOCUMENTS[origin](ts) + compression

DIFFS_ENTITIES = DateTimeKey(DIFFS / ENTITIES_JSON, TS_FORMAT, EXT_ENTITIES_DELTA) module-attribute

Entities diff series: DIFFS_ENTITIES(ts) + compression

ENTITIES_JSON = StoreKey('entities.ftm.json') module-attribute

aggregated entities export – the identity; + compression for the artifact

EXPORTS = StoreKey('exports') module-attribute

Base path for exports

EXPORTS_CYPHER = EXPORTS / 'graph.cypher' module-attribute

neo4j data export file path

EXPORTS_DOCUMENTS = EXPORTS / 'documents.csv' module-attribute

documents metadata export – codec-free, unscoped

EXPORTS_STATEMENTS = EXPORTS / 'statements.csv' module-attribute

complete sorted statements export – codec-free

EXPORTS_STATISTICS = EXPORTS / STATISTICS module-attribute

entity counts, pre-computed facts file path

EXT_DOCUMENTS_DELTA = 'diff.csv' module-attribute

Extension of one documents diff file

EXT_ENTITIES_DELTA = 'delta.json' module-attribute

Extension of one entities diff file

INDEX = StoreKey('index.json') module-attribute

generated index filename

JOBS = StoreKey('jobs') module-attribute

Job data prefix

JOB_RUNS = JobsKey(JOBS / 'runs') module-attribute

Job runs result storage prefix, and the factory for one run

LOCK = StoreKey('.LOCK') module-attribute

dataset-wide maintenance lock key name

LOCKS = ScopedKey('.locks', TENANT) module-attribute

Locks, under the default tenant: .locks/lakehouse/. LOCKS["other"] for another tenant.

LOCK_APPENDS = StoreKey('.LOCK-APPENDS') module-attribute

Prefix for per-writer append marker keys (shared side of the write fence)

STATEMENTS = 'statements' module-attribute

Base path for storing statement data (partitioned by shard, bucket, origin)

STATISTICS = StoreKey('statistics.json') module-attribute

computed statistics filename

TAGS = ScopedKey('tags', TENANT) module-attribute

Freshness tags, under the default tenant: tags/lakehouse/ TAGS["other"] for another tenant.

TENANT = 'lakehouse' module-attribute

Default tenant name

TS_FORMAT = '%Y%m%dT%H%M%S%fZ' module-attribute

Global format for timestamps in files

VERSIONS = VersionsKey() module-attribute

Base path for versions, and the factory for one snapshot

ArchiveKey

Bases: StoreKey

The directory holding one archived file, and the files in it.

Layout: archive/5a/6a/cf/5a6acf229ba576d9a40b09292595658bbb74ef56/

One checksum is stored once, but it can have arrived by several source paths and been read by several text extractors – hence meta and txt being keyed rather than fixed like blob.

Parameters:

Name Type Description Default
checksum str

SHA256 checksum of the file

required
Source code in ftm_lakehouse/core/conventions/path.py
class ArchiveKey(StoreKey):
    """The directory holding one archived file, and the files in it.

    Layout: ``archive/5a/6a/cf/5a6acf229ba576d9a40b09292595658bbb74ef56/``

    One checksum is stored once, but it can have arrived by several source
    paths and been read by several text extractors – hence
    [`meta`][ArchiveKey.meta] and [`txt`][ArchiveKey.txt] being keyed rather
    than fixed like [`blob`][ArchiveKey.blob].

    Args:
        checksum: SHA256 checksum of the file
    """

    def __init__(self, checksum: str) -> None:
        super().__init__(ARCHIVE, make_checksum_key(checksum))

    @property
    def blob(self) -> StoreKey:
        """The file's content, stored once per checksum."""
        return self / ARCHIVE_BLOB

    def meta(self, file_id: str) -> StoreKey:
        """Metadata for one file instance.

        Several files with the same checksum but different source paths each
        get their own metadata, keyed by their ``File.id``.

        Layout: ``archive/5a/6a/cf/.../file-abc123.json``

        Args:
            file_id: The ``File.id`` (hash of source path + checksum)

        Raises:
            ValueError: If ``file_id`` is malformed.
        """
        return self / f"{safe_name(file_id, 'file_id')}.json"

    def txt(self, origin: str) -> StoreKey:
        """Extracted text for one extraction origin.

        Several extractions can exist per file, keyed by origin (different OCR
        engines or extraction methods).

        Layout: ``archive/5a/6a/cf/.../{origin}.txt``

        Args:
            origin: The extraction origin / engine name

        Raises:
            ValueError: If ``origin`` is malformed.
        """
        return self / f"{validate_origin(origin)}.txt"

blob property

The file's content, stored once per checksum.

meta(file_id)

Metadata for one file instance.

Several files with the same checksum but different source paths each get their own metadata, keyed by their File.id.

Layout: archive/5a/6a/cf/.../file-abc123.json

Parameters:

Name Type Description Default
file_id str

The File.id (hash of source path + checksum)

required

Raises:

Type Description
ValueError

If file_id is malformed.

Source code in ftm_lakehouse/core/conventions/path.py
def meta(self, file_id: str) -> StoreKey:
    """Metadata for one file instance.

    Several files with the same checksum but different source paths each
    get their own metadata, keyed by their ``File.id``.

    Layout: ``archive/5a/6a/cf/.../file-abc123.json``

    Args:
        file_id: The ``File.id`` (hash of source path + checksum)

    Raises:
        ValueError: If ``file_id`` is malformed.
    """
    return self / f"{safe_name(file_id, 'file_id')}.json"

txt(origin)

Extracted text for one extraction origin.

Several extractions can exist per file, keyed by origin (different OCR engines or extraction methods).

Layout: archive/5a/6a/cf/.../{origin}.txt

Parameters:

Name Type Description Default
origin str

The extraction origin / engine name

required

Raises:

Type Description
ValueError

If origin is malformed.

Source code in ftm_lakehouse/core/conventions/path.py
def txt(self, origin: str) -> StoreKey:
    """Extracted text for one extraction origin.

    Several extractions can exist per file, keyed by origin (different OCR
    engines or extraction methods).

    Layout: ``archive/5a/6a/cf/.../{origin}.txt``

    Args:
        origin: The extraction origin / engine name

    Raises:
        ValueError: If ``origin`` is malformed.
    """
    return self / f"{validate_origin(origin)}.txt"

VersionsKey

Bases: CallableKey

versions/: a prefix to iterate, and a factory for one snapshot.

Source code in ftm_lakehouse/core/conventions/path.py
class VersionsKey(CallableKey):
    """``versions/``: a prefix to iterate, and a factory for one snapshot."""

    def __init__(self) -> None:
        super().__init__("versions")

    def __call__(self, name: str, ts: datetime | str | None = None) -> StoreKey:
        """Get a versioned snapshot path for a file (``index.json``, ``config.yml``).

        Layout: ``versions/YYYY/MM/{TS_FORMAT}/<name>``

        Args:
            name: The file name to version (e.g. ``config.yml``, ``index.json``)
            ts: Timestamp of the snapshot, omit to use current time

        Returns:
            Key like ``versions/2025/01/20250115T103000000000Z/config.yml``
        """
        if not isinstance(ts, str):
            ts = make_ts(ts, TS_FORMAT)
        return self / ts[:4] / ts[4:6] / ts / name

__call__(name, ts=None)

Get a versioned snapshot path for a file (index.json, config.yml).

Layout: versions/YYYY/MM/{TS_FORMAT}/<name>

Parameters:

Name Type Description Default
name str

The file name to version (e.g. config.yml, index.json)

required
ts datetime | str | None

Timestamp of the snapshot, omit to use current time

None

Returns:

Type Description
StoreKey

Key like versions/2025/01/20250115T103000000000Z/config.yml

Source code in ftm_lakehouse/core/conventions/path.py
def __call__(self, name: str, ts: datetime | str | None = None) -> StoreKey:
    """Get a versioned snapshot path for a file (``index.json``, ``config.yml``).

    Layout: ``versions/YYYY/MM/{TS_FORMAT}/<name>``

    Args:
        name: The file name to version (e.g. ``config.yml``, ``index.json``)
        ts: Timestamp of the snapshot, omit to use current time

    Returns:
        Key like ``versions/2025/01/20250115T103000000000Z/config.yml``
    """
    if not isinstance(ts, str):
        ts = make_ts(ts, TS_FORMAT)
    return self / ts[:4] / ts[4:6] / ts / name

Tag conventions

Global tags used to identify actions. Used for cache keys of workflow runs etc.

Export operations don't have constants here – their freshness tag is the path.* export target itself (e.g. exports/statements.csv), touched by DatasetJobOperation._run_local after a successful run.

ARCHIVE_ORIGIN = 'archive' module-attribute

Default origin identifier for archived files (if not crawled)

ARCHIVE_UPDATED = 'archive/last_updated' module-attribute

Archive last updated (file added or removed)

CRAWL_ORIGIN = 'crawl' module-attribute

Default origin identifier for crawled files.

JOURNAL_FLUSHED = 'journal/last_flushed' module-attribute

Journal store last flushed into statement store

JOURNAL_UPDATED = 'journal/last_updated' module-attribute

Statement journal was updated

OP_CRAWL = 'operations/crawl/last_run' module-attribute

Last crawl (import files) execution

OP_DOWNLOAD_ARCHIVE = 'operations/download_archive/last_run' module-attribute

Last download archive execution

OP_EXPORT = 'operations/export/last_run' module-attribute

Last fused export sweep (ExportKind.all).

The individual artifacts keep their own freshness tags – the sweep stamps every one it writes – so this is the tag for "the sweep as a whole ran", which is what a subsequent export all checks itself against.

OP_MAKE = 'operations/make/last_run' module-attribute

Last make (full workflow) execution

OP_MIGRATE = 'operations/migrate/last_run' module-attribute

Last migrate (outstanding dataset migrations applied)

OP_SHARD = 'operations/shard/last_run' module-attribute

Last re-shard (statement store rewritten onto a new shard count)

STATEMENTS_OPTIMIZED = 'statements/last_optimized' module-attribute

Statement store was optimized (merge + compact + vacuum)

STATEMENTS_UPDATED = 'statements/last_updated' module-attribute

Statement store was updated

migration(name)

Applied-marker tag for a single migration.

Presence, not recency, is the state: a dataset carrying this tag has run that migration and MigrateOperation skips it.

Parameters:

Name Type Description Default
name str

Name of a migration function in ftm_lakehouse.operation.migrations.

required
Source code in ftm_lakehouse/core/conventions/tag.py
def migration(name: str) -> str:
    """Applied-marker tag for a single migration.

    Presence, not recency, is the state: a dataset carrying this tag has run
    that migration and
    [`MigrateOperation`][ftm_lakehouse.operation.maintenance.MigrateOperation]
    skips it.

    Args:
        name: Name of a migration function in
            ``ftm_lakehouse.operation.migrations``.
    """
    return f"migrations/{name}"

statements_partition_optimized(shard, bucket, origin)

Per-partition freshness tag: a (shard, bucket, origin) was merged.

Partition-scoped analog of STATEMENTS_OPTIMIZED, stamped by ParquetStore.merge after it rewrites the partition. See statements_partition_updated for the freshness comparison.

Parameters:

Name Type Description Default
shard str

Hex-padded shard value.

required
bucket str

FtM schema bucket (thing / interval / ...).

required
origin str

Source tag – validated so it stays a single path segment.

required
Source code in ftm_lakehouse/core/conventions/tag.py
def statements_partition_optimized(shard: str, bucket: str, origin: str) -> str:
    """Per-partition freshness tag: a ``(shard, bucket, origin)`` was merged.

    Partition-scoped analog of [`STATEMENTS_OPTIMIZED`][STATEMENTS_OPTIMIZED],
    stamped by
    [`ParquetStore.merge`][ftm_lakehouse.storage.parquet.ParquetStore.merge]
    after it rewrites the partition. See
    [`statements_partition_updated`][statements_partition_updated] for the
    freshness comparison.

    Args:
        shard: Hex-padded shard value.
        bucket: FtM schema bucket (``thing`` / ``interval`` / ...).
        origin: Source tag – validated so it stays a single path segment.
    """
    validate_origin(origin)
    return f"statements/{shard}/{bucket}/{origin}/last_optimized"

statements_partition_updated(shard, bucket, origin)

Per-partition freshness tag: a (shard, bucket, origin) was written.

Partition-scoped analog of STATEMENTS_UPDATED, stamped by ParquetStore.append. ParquetStore.merge compares it against statements_partition_optimized via TagStore.is_latest to skip partitions that haven't changed since their last merge.

Parameters:

Name Type Description Default
shard str

Hex-padded shard value.

required
bucket str

FtM schema bucket (thing / interval / ...).

required
origin str

Source tag – validated so it stays a single path segment.

required
Source code in ftm_lakehouse/core/conventions/tag.py
def statements_partition_updated(shard: str, bucket: str, origin: str) -> str:
    """Per-partition freshness tag: a ``(shard, bucket, origin)`` was written.

    Partition-scoped analog of [`STATEMENTS_UPDATED`][STATEMENTS_UPDATED],
    stamped by
    [`ParquetStore.append`][ftm_lakehouse.storage.parquet.ParquetStore.append].
    [`ParquetStore.merge`][ftm_lakehouse.storage.parquet.ParquetStore.merge]
    compares it against
    [`statements_partition_optimized`][statements_partition_optimized] via
    [`TagStore.is_latest`][ftm_lakehouse.storage.tags.TagStore.is_latest] to
    skip partitions that haven't changed since their last merge.

    Args:
        shard: Hex-padded shard value.
        bucket: FtM schema bucket (``thing`` / ``interval`` / ...).
        origin: Source tag – validated so it stays a single path segment.
    """
    validate_origin(origin)
    return f"statements/{shard}/{bucket}/{origin}/last_updated"