Build a mapping step

A mapping step merges equivalent keys from multiple upstream runs, writes a canonical registry, and writes a source-run/local-id crosswalk. The merge algorithm belongs to framework mapping.

1. Declare upstream runs

params:
  mapping:
    upstream_runs: ""
    run_id: ""

Expose mapping.upstream_runs as a runs picker in ui.streamlit when launched interactively.

2. Declare datasets and hash merge

drtml: 4
step: merge_tokens
run:
  type: mapping
  entities: token_registry_run
  registry: token_registry_canonical
  crosswalk: token_mapping
  ids:
    registry: token_id
    crosswalk: token_id
  source_run: source_run_id
orchestration:
  artifacts_root: env:DRTOLLER_ARTIFACTS_ROOT
storage:
  default: artifacts
  stores:
    artifacts: {backend: s3, root: pipeline-artifacts, profile: minio_app}
inputs:
  token_registry_run:
    run: mapping.upstream_runs
    identity:
      method: hash
      by:
        language: string
        token: string
      source: token_id_run
      hash: token_id
outputs:
  token_registry_canonical:
    schema:
      token_id: string
      language: string
      token: string
  token_mapping:
    schema:
      source_run_id: string
      token_id_run: string
      token_id: string

3. Thin hook (optional)

from drtoller.framework.drtml.resolve import manifest_path_adjacent
from drtoller.framework.processing.mapping_mode.run import execute_mapping_build


def run(ctx, *, manifest_path=None):
    path = manifest_path or manifest_path_adjacent(__file__, step_id="merge_tokens")
    return execute_mapping_build(ctx, manifest_path=path)

The step contains no merge algorithm. Downstream steps load the crosswalk through lookups / run.lookup_bindings and remap ids in domain code.