Skip to content

Conventions

The path layout, artifact names and freshness tags are stable contracts: third-party tools can populate or consume a lakehouse by following them, without this library.

Storage Layout

Layout of a lakehouse storage root, local or remote:

lakehouse/
└── {dataset}/
    ├── config.yml                # Dataset configuration (shards, compression, ...)
    ├── index.json                # Published dataset index with statistics
    ├── .LOCK                     # Exclusive maintenance fence
    ├── .LOCK-MERGE               # Merge / export-sweep lock
    │
    ├── archive/                  # Content-addressed file storage
    │   └── {ch[0:2]}/{ch[2:4]}/{ch[4:6]}/{checksum}/
    │       ├── blob              # Raw file content (stored once)
    │       └── {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]  # Statements export
    │   ├── statistics.json           # Entity counts, facets
    │   ├── parents.csv[.gz|.zst]     # Folders documents sit in, with their paths
    │   ├── documents.csv[.gz|.zst]   # Document metadata
    │   └── documents.{origin}.csv[..] # Document metadata, one origin only
    │
    ├── diffs/                    # Timestamped diff exports
    ├── versions/                 # Versioned snapshots (config, index, ...)
    │   └── YYYY/MM/{timestamp}/
    ├── tags/{tenant}/            # Freshness tags (default tenant: lakehouse)
    └── jobs/
        └── runs/{job_type}/{timestamp}.json

With LAKEHOUSE_STATEMENTS_URI set, statements/ sits at {LAKEHOUSE_STATEMENTS_URI}/{dataset}/statements instead; everything else stays under the storage root – see Separate statements storage.

Freshness Tags

Operations skip work their tags say is fresh: is_latest(key, dependencies) is True when key is newer than all its dependencies.

Tag Set by Meaning
statements/last_updated Flush / append, delete_origin The store's content moved – what exports, statistics and diffs depend on. A merge rewrites files, not content, and leaves it alone
archive/last_updated Archiving a file A file was archived
exports/statements.csv, entities.ftm.json, exports/parents.csv, exports/documents.csv, exports/documents.{origin}.csv, exports/statistics.json, index.json export Each artifact's key is its tag, stamped when a run wrote it. archive download (DownloadArchiveOperation) depends on exports/documents.csv
operations/{name}/last_run crawl, make, export, optimize, shard, migrate, download_archive When the operation last completed. optimize decides freshness from the store's dirty partitions, not from its tag
migrations/{function} maintenance migrate One per applied migration

Compression suffixes

With compression set in config.yml (gz / zst), the streamed artifacts carry the codec suffix – entities.ftm.json.zst, exports/statements.csv.zst, exports/parents.csv.zst, exports/documents.csv.zst – and index.json lists those names and urls. index.json and statistics.json are always plain JSON.

Diff directories stay codec-free (diffs/exports/documents.csv/) – they are named after the tag; only the files in them carry the suffix ({timestamp}.diff.csv.zst).

Path conventions

Every path is built through ftm_lakehouse.core.conventions.path:

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-MERGE                     # merge / export-sweep lock
        .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)
                {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_PARENTS = EXPORTS / 'parents.csv' module-attribute

every folder a document can sit in, with its path – codec-free

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

complete 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_MERGE = StoreKey('.LOCK-MERGE') module-attribute

Merge lock – held by merge and by an export sweep, and taken alongside .LOCK by the exclusive maintenance. Appends do not wait for it.

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 been read by several text extractors – hence 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 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
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 been read by several text
    extractors – hence [`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 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.

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 
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
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 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
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 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
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)

CRAWL_ORIGIN = 'crawl' module-attribute

Default origin identifier for crawled files.

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 export run.

The individual artifacts keep their own freshness tags – the run stamps every one it writes – so this is the tag for "the export as a whole ran", and what the next export 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_OPTIMIZE = 'operations/optimize/last_run' module-attribute

Last optimize (merge + vacuum) execution – stamped by the run, never read for freshness: OptimizeOperation asks the statement store for dirty partitions instead.

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

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

STATEMENTS_UPDATED = 'statements/last_updated' module-attribute

The statement store's content moved – rows appended (a flush) or an origin dropped. The one clock every export, statistic and diff depends on: a merge rewrites files but changes no content, so it does not touch it.

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 
45
46
47
48
49
50
51
52
53
54
55
56
57
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}"