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
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 |
required |
Raises:
| Type | Description |
|---|---|
ValueError
|
If |
Source code in ftm_lakehouse/core/conventions/path.py
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 |
Source code in ftm_lakehouse/core/conventions/path.py
VersionsKey
Bases: CallableKey
versions/: a prefix to iterate, and a factory for one snapshot.
Source code in ftm_lakehouse/core/conventions/path.py
__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. |
required |
ts
|
datetime | str | None
|
Timestamp of the snapshot, omit to use current time |
None
|
Returns:
| Type | Description |
|---|---|
StoreKey
|
Key like |
Source code in ftm_lakehouse/core/conventions/path.py
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
|
required |
Source code in ftm_lakehouse/core/conventions/tag.py
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 ( |
required |
origin
|
str
|
Source tag – validated so it stays a single path segment. |
required |
Source code in ftm_lakehouse/core/conventions/tag.py
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 ( |
required |
origin
|
str
|
Source tag – validated so it stays a single path segment. |
required |