Lineage observation hub

Apache Hop provides an in-engine lineage observation hub for recording metadata-level observations (pipeline context, file I/O, HTTP traffic, structural hints) and delivering them asynchronously to pluggable lineage sinks. The design keeps third-party integrations (OpenLineage, Datadog, custom exporters) out of the engine: sinks are separate plugins that map Hop’s neutral events to their own formats.

The public API lives under the Java package org.apache.hop.lineage in the hop-engine module.

Goals

  • Central entry point — Producers call LineageHub.getInstance().emit(…​) instead of talking to each backend directly.

  • Safe defaults — The hub is off until HOP_LINEAGE_ENABLED is set to Y.

  • Non-blocking pipelines — Events are queued and dispatched on a dedicated worker thread; sinks run in batch with per-sink error isolation.

  • Extensibility — New backends ship as ILineageSink implementations registered via @LineageSinkPlugin.

Package layout

Package Role

org.apache.hop.lineage

LineageVariables — Hop variables (@Variable) that describe hub configuration. Also the per-kind emitters (LineageFileIoEmitter, LineageRelationalIoEmitter, LineageHttpIoEmitter, LineageRunLifecycleEmitter, LineageTransformSchemaEmitter), the shared dataset-identity helpers (LineageFileIdentity, LineageRelationalIdentity) and LineageMetadataWalker (see Relational lineage is declared, not emitted). Emitters are best-effort by construction: they no-op on missing arguments and never throw into transform execution, so call sites invoke them directly without a defensive try/catch.

org.apache.hop.lineage.context

LineageContext, LineageSubjectType, LineagePortableFilename — Immutable correlation (subject type, log channel, pipeline/workflow/transform/action names, hopFilename as on the engine/meta, optional hopFilenamePortableKey as ${PROJECT_HOME}/… when derivable, optional attributes). Run identity for a single execution is the log channel id on the context.

org.apache.hop.lineage.model

LineageEvent, LineageEventKind, LineagePayload and concrete payloads (structural, file I/O, relational I/O, HTTP, run lifecycle).

org.apache.hop.lineage.hub

LineageHub, LineageConfiguration — Queue, batching, flush, shutdown.

org.apache.hop.lineage.spi

ILineageSink — Plugin SPI.

org.apache.hop.lineage.plugin

LineageSinkPlugin, LineageSinkPluginType — Plugin type registered in HopEnvironment.

org.apache.hop.lineage.xp

Extension points that call LineageHub.flushQuietly() on pipeline/workflow/server lifecycle (see [lineage-flush]).

Relational lineage is declared, not emitted

A transform does not emit RELATIONAL_IO events itself. It declares the table access its metadata describes, and the engine derives the lineage from that declaration when the transform finishes.

Two annotations carry the declaration:

  • @RelationalLineage (org.apache.hop.lineage.api) on the ITransformMeta class — opts the transform in and states the RelationalIoOperation (WRITE, DELETE, …). Opting in is explicit rather than inferred from the presence of a table property, because plenty of metadata names a table without reading or writing its rows (checking that it exists, comparing two of them).

  • HopMetadataPropertyType values on the properties themselves — RDBMS_CONNECTION, RDBMS_SCHEMA, RDBMS_TABLE, RDBMS_TRUNCATE, and RDBMS_COLUMN paired with the database-neutral STREAM_FIELD on the same mapping object to express "stream field X is written to column Y". These annotations already existed for search and metadata injection; lineage reuses them.

LineageMetadataWalker reads that declaration, recursing into nested metadata objects and lists. LineageRelationalIoEmitter.emitDeclaredRelationalIo then combines it with the facts only available at runtime — resolved variables, the connection the name refers to, the row shape, and the transform that produced each stream field — and emits the event. LineageHubTransformFinishXp drives this for every transform.

Consequences worth knowing when adding a database transform:

  • Annotating the metadata is all that is required; there is no emission code to write, and no way to silently forget it for a transform that already annotates its properties.

  • A column declaring RDBMS_COLUMN with no STREAM_FIELD is treated as generated by the transform (a dimension’s technical key, for example) and is left out of the column lineage rather than reported with no origin.

  • Facts that only exist per row cannot be declared. A target table read from a row field, or a SQL statement built at runtime, is still reported by the transform itself — Table Output’s dynamic table-name mode and the SQL-driven transforms (Table Input, Execute SQL) work that way.

  • A transform reports its table access when it finishes, which is after the pipeline run’s own completion event. Marquez creates dataset versions only from terminal run events, so such a dataset shows fields: [] and no current version in its API and UI, while its columnLineage facet is stored and the lineage graph is complete. This is Marquez’s ingestion rule, not a gap in what Hop emits — worth knowing before investigating a dataset that looks empty.

Event model

Each observation is a LineageEvent:

  • Event id — Unique id (UUID by default).

  • Timestamp — Milliseconds since epoch.

  • LineageEventKind — One of RUN_LIFECYCLE, TRANSFORM_SCHEMA, FILE_IO, RELATIONAL_IO, HTTP_IO.

  • LineageContext — Who/what produced the observation (see above).

  • LineagePayload — Optional typed payload; may be null if the kind and context are enough.

Payload types are plain Java types (strings, longs, enums) so sinks can serialize them without dragging engine objects across boundaries:

  • FileIoLineagePayload — Operation (READ, WRITE, COPY, MOVE, DELETE), source/target URIs, byte counts, success, message, and optional FileIoContentSchema: tabular columns actually read/written (including JSONPath / YAML path / XPath locators) plus optional structureRoots (FileIoSchemaNode tree) merged from those paths for nested file layouts. Text File Output and JSON Output attach the file column layout (not the full transform output row when a subset is written).

  • RelationalIoLineagePayload — Relational/JDBC table access: operation (READ, WRITE, DELETE, EXEC), the resolved dataset namespace, the input/output RelationalTable list (catalog/schema/table), the raw SQL when the engine cannot resolve the tables itself, an optional column schema, the connection’s default catalog and schema, optional per-column provenance (RelationalWriteColumn: target column, stream field, origin transform), and an optional RelationalLifecycle (OVERWRITE, TRUNCATE, DROP, CREATE).

  • HttpLineagePayload — CLIENT vs SERVER, method, URL, status, request/response bytes, duration, success, message.

  • RunLifecycleLineagePayload — STARTED / FINISHED plus optional detail text.

  • TransformSchemaLineagePayload — INPUT or OUTPUT plus a list of field names, Hop type names, length, and precision (boundary schema; emitted when a transform finishes). Fields usually come from runtime row metadata on the transform or row sets. If input was never observed (e.g. Abort when getRow() immediately returns null), input is derived via PipelineMeta#getPrevTransformFields. If output was never written (e.g. Detect empty stream with a non-empty stream), output is derived via PipelineMeta#getTransformFields. Graph-derived events set context attribute transformSchemaSource to DESIGN_GRAPH.

Convenience factory: LineageEvent.of(kind, context, payload) assigns id and current time.

Hub behavior

LineageHub is a singleton (getInstance()). When lineage is disabled, emit returns immediately.

When enabled:

  1. The first emit starts a daemon worker thread (Hop-LineageHub-Dispatcher).

  2. Events are placed on a bounded queue. If the queue is full, the event is dropped and an error is logged (drops are counted).

  3. The worker builds batches up to HOP_LINEAGE_BATCH_MAX, waiting up to HOP_LINEAGE_BATCH_LINGER_MS for additional events before sending a partial batch.

  4. Each batch is passed to every loaded sink via ILineageSink.accept(List<LineageEvent>). If one sink throws, others still receive the same batch; failures are logged.

flush() blocks until queued events (and coalesced flush markers) have been processed. flushQuietly() catches and logs errors.

shutdown() stops the worker and calls shutdown() on sinks. It is invoked from HopEnvironment.shutdown() and HopEnvironment.reset().

After PluginRegistry.init(), HopEnvironment calls LineageHub.getInstance().environmentReady() so sink lists can be rebuilt when the environment is re-initialized.

The queue is allocated lazily on the first emit from the resolved configuration, so HOP_LINEAGE_QUEUE_CAPACITY applies to the singleton as well as to isolated test hubs. The capacity is fixed for the lifetime of the queue; to change it, restart the JVM (or call shutdown() and emit again in tests).

Run lifecycle (extension points)

The org.apache.hop.lineage.xp package registers extension points that emit RUN_LIFECYCLE events (via LineageRunLifecycleEmitter) and flush the hub where needed:

  • PipelineStart — LineageHubPipelineStartXp — phase STARTED

  • PipelineFinish — LineageHubPipelineFinishXp — phase FINISHING (transforms stopped, cleanup)

  • PipelineCompleted — LineageHubPipelineCompletedXp — phase FINISHED or FAILED (from pipeline error count), then flushQuietly()

  • WorkflowStart — LineageHubWorkflowStartXp — phase STARTED

  • WorkflowFinish — LineageHubWorkflowFinishXp — phase FINISHED or FAILED (from workflow/result errors), then flushQuietly()

  • HopServerShutdown — LineageHubServerShutdownFlushXp — flush only

  • TransformBeforeStart — LineageHubTransformStartXp — transform phase STARTED (local pipeline engine; transform log channel + pipeline correlation attributes)

  • TransformFinished — LineageHubTransformFinishXp — transform FINISHED or FAILED from transform error count, then TRANSFORM_SCHEMA events for observed input/output row shapes when non-empty

  • WorkflowBeforeActionExecution — LineageHubWorkflowActionBeforeXp — action STARTED (workflow log channel + action name; template action log channel id as attribute when present)

  • WorkflowAfterActionExecution — LineageHubWorkflowActionAfterXp — action FINISHED or FAILED from the action Result; uses the executed action’s log channel id when available (WorkflowExecutionExtension.actionLogChannelId / actionExecutionResult)

Remote (RemotePipelineEngine) and Beam (BeamPipelineEngine) engines also call PipelineStart and PipelineFinish so they align with the local pipeline engine.

Beam and other engines that do not use the classic Hop thread-per-transform model typically do not fire TransformBeforeStart / TransformFinished the same way as the local Pipeline engine.

Configuration

Variables are defined on org.apache.hop.lineage.LineageVariables and appear in the variable registry (ENGINE scope) like other Hop settings.

Variable Default Meaning

HOP_LINEAGE_ENABLED

N

Set to Y to enable the hub.

HOP_LINEAGE_QUEUE_CAPACITY

10000

Bounded queue capacity. Read on first emit; the queue retains this size for its lifetime.

HOP_LINEAGE_BATCH_MAX

100

Maximum events per batch per sink.

HOP_LINEAGE_BATCH_LINGER_MS

250

How long the worker waits for more events before sending a smaller batch.

HOP_LINEAGE_SINK_IDS

(empty)

Comma-separated sink plugin ids (case-insensitive). Empty means all discovered sinks are loaded.

LineageConfiguration.resolve() reads the active IVariables (typically after Variables.initializeFrom so system properties and Hop config apply).

LineageSink plugins

Sinks implement org.apache.hop.lineage.spi.ILineageSink:

  • init(IVariables variables, ILogChannel log) — Optional; called once before the first batch.

  • accept(List<LineageEvent> events) — Receives a non-empty batch; should not hold references to the list after returning.

  • shutdown() — Optional cleanup.

The implementation class is annotated with @LineageSinkPlugin (id, optional name, description), mirroring other Hop plugin types (see Plugin Development).

The plugin type LineageSinkPluginType is registered in HopEnvironment.getStandardPluginTypes() alongside transforms, actions, execution info locations, etc. Plugins are discovered from the classpath like other native/engine plugins.

No sink implementations ship with the engine. Integrators implement ILineageSink, annotate the class with @LineageSinkPlugin(id = "…​"), and package it as a Hop plugin. With no sinks loaded, the hub remains a no-op even when HOP_LINEAGE_ENABLED=Y.

Emitting events from code

Producers (pipeline/workflow hooks, VFS usage, HTTP client/server filters, transforms) should build a LineageContext (often via LineageContext.builder()), optionally attach a payload, then emit:

// IVariables vars = pipeline; // or workflow — must expose PROJECT_HOME when normalizing
LineageEvent event =
    LineageEvent.of(
        LineageEventKind.FILE_IO,
        LineageContext.builder()
            .subjectType(LineageSubjectType.ACTION)
            .hopFilename(filename)
            .hopFilenamePortableKey(
                LineagePortableFilename.portableKey(filename, vars))
            .build(),
        new FileIoLineagePayload(
            FileIoOperation.COPY, sourceUri, targetUri, bytes, true, null));
LineageHub.getInstance().emit(event);

Keep call sites thin: one object construction plus emit. Heavy work belongs in sinks or async code inside the sink implementation.

Related topics