Package {openlineage}

Create and Emit OpenLineage Events

Description

Build validated OpenLineage 2.0.2 run events, serialize them to the protocol wire format, and emit them through local or synchronous HTTP transports.

Main workflow

Create a Run, Job, and optional input/output datasets, then combine them in a RunEvent. An OpenLineageClient emits the event through an injected transport or through HTTP configured by constructor arguments or environment variables.

Facets and extensions

Typed facet constructors cover common metadata. Use ol_facet to attach a facet schema that does not yet have a typed R constructor.

Author(s)

Maintainer: Pedro Baltazar pedrobtz@gmail.com [copyright holder]

Authors:

• Pedro Baltazar pedrobtz@gmail.com [copyright holder]

See Also

OpenLineageClient, RunEvent, local_transports, HttpTransport

Synchronous HTTP transport

Description

Sends OpenLineage events as JSON POST requests. The base url and relative endpoint are joined with one slash; a URL that already ends in the endpoint is not duplicated.

Supply api_key for ⁠Authorization: Bearer <key>⁠ authentication, or place authentication headers in headers. Supplying both an API key and a custom Authorization header is an error. All custom headers are redacted from httr2 request printing.

HTTP 429, 500, 502, 503, and 504 responses and low-level connection failures are retried up to max_retries times. HTTP 401 and 403 responses are never retried. Set verify_tls = FALSE only for controlled development environments; doing so raises an openlineage_insecure_tls_warning.

Value

An R6 HTTP transport object.

Super class

OpenLineageTransport -> HttpTransport

Methods

Public methods

• HttpTransport$new()

• HttpTransport$emit()

• HttpTransport$print()

• HttpTransport$clone()

HttpTransport$new()

Create an HTTP transport.

Usage

HttpTransport$new(
  url,
  endpoint = "api/v1/lineage",
  api_key = NULL,
  headers = character(),
  timeout = 5,
  verify_tls = TRUE,
  max_retries = 3L
)

Arguments

Returns

A new HttpTransport object.

HttpTransport$emit()

Send an event to the configured OpenLineage HTTP endpoint.

Usage

HttpTransport$emit(event)

Arguments

Returns

event, invisibly, after a successful HTTP response.

HttpTransport$print()

Print a redacted transport summary.

Usage

HttpTransport$print(...)

Arguments

Returns

The transport, invisibly.

HttpTransport$clone()

The objects of this class are cloneable with this method.

Usage

HttpTransport$clone(deep = FALSE)

Arguments

See Also

local_transports

Examples

transport <- HttpTransport$new("https://example.com")
transport

OpenLineage job

Description

Identifies a job within a producer-defined namespace.

Usage

Job(namespace, name, facets = list())

Arguments

Value

A Job S7 object.

Examples

Job("example-scheduler", "daily-report")

OpenLineage client

Description

Coordinates event emission through an OpenLineage transport. Supply a transport directly for local or custom delivery, or configure the built-in HTTP transport with url and related arguments.

When an argument is absent, the client reads OPENLINEAGE_URL, OPENLINEAGE_ENDPOINT, OPENLINEAGE_API_KEY, and OPENLINEAGE_DISABLED. Explicit arguments take precedence over their environment equivalents. OPENLINEAGE_DISABLED accepts only true or false, ignoring case and surrounding whitespace.

A disabled client uses NoopTransport. An injected transport takes precedence over HTTP environment variables. If neither a transport nor URL is configured, the client uses ConsoleTransport. Supplying partial HTTP settings without a URL is an error.

Value

An R6 OpenLineage client object.

Public fields

Methods

Public methods

• OpenLineageClient$new()

• OpenLineageClient$emit()

• OpenLineageClient$print()

• OpenLineageClient$clone()

OpenLineageClient$new()

Create an OpenLineage client.

Usage

OpenLineageClient$new(
  transport = NULL,
  url = NULL,
  endpoint = "api/v1/lineage",
  api_key = NULL,
  headers = character(),
  timeout = 5,
  verify_tls = TRUE,
  max_retries = 3L,
  disabled = NULL
)

Arguments

Returns

A new OpenLineageClient object.

OpenLineageClient$emit()

Emit an OpenLineage event through the configured transport.

Usage

OpenLineageClient$emit(event)

Arguments

Returns

event, invisibly, after successful delivery.

OpenLineageClient$print()

Print a credential-free client summary.

Usage

OpenLineageClient$print(...)

Arguments

Returns

The client, invisibly.

OpenLineageClient$clone()

The objects of this class are cloneable with this method.

Usage

OpenLineageClient$clone(deep = FALSE)

Arguments

See Also

local_transports, HttpTransport

Examples

transport <- AccumulatingTransport$new()
client <- OpenLineageClient$new(transport = transport, disabled = FALSE)
event <- RunEvent(
  Run(new_run_id()),
  Job("example", "task"),
  event_type = "START"
)
client$emit(event)
length(transport$events)

OpenLineage run

Description

Identifies one execution of an OpenLineage job.

Usage

Run(run_id, facets = list())

Arguments

Value

A Run S7 object.

Examples

Run(new_run_id())

OpenLineage run event

Description

Describes a lifecycle transition for one job run and its input and output datasets.

Usage

RunEvent(
  run,
  job,
  event_type = NULL,
  event_time = new_event_time(),
  inputs = list(),
  outputs = list(),
  producer = OPENLINEAGE_PRODUCER,
  schema_url = OPENLINEAGE_RUN_EVENT_SCHEMA_URL
)

Arguments

Value

A RunEvent S7 object.

Examples

run <- Run(new_run_id())
job <- Job("example-scheduler", "daily-report")
RunEvent(
  run,
  job,
  event_type = "START",
  event_time = new_event_time(as.POSIXct("2026-01-02", tz = "UTC"))
)

OpenLineage run state

Description

A validated lifecycle state for an OpenLineage run.

Usage

RunState(value)

Arguments

Value

A RunState S7 object.

Examples

RunState("START")

Convert an OpenLineage value to its wire representation

Description

Recursively converts OpenLineage S7 models to named R lists. Protocol field names are applied here, RunState values are unwrapped, and NULL properties are omitted. Object key order is deterministic; array order is preserved.

Usage

as_openlineage_list(x)

Arguments

Value

A named list for a model, or the corresponding wire value for a nested value.

Examples

run <- Run("019c0000-0000-7000-8000-000000000001")
as_openlineage_list(run)

Typed dataset facets

Description

Constructors for dataset-level schema, data-source, and tag metadata.

Usage

SchemaField(
  name,
  type = NULL,
  description = NULL,
  ordinal_position = NULL,
  fields = list()
)

SchemaDatasetFacet(
  fields = list(),
  producer = OPENLINEAGE_PRODUCER,
  deleted = NULL
)

DatasourceDatasetFacet(
  name = NULL,
  uri = NULL,
  producer = OPENLINEAGE_PRODUCER,
  deleted = NULL
)

Arguments

Value

A typed dataset facet or schema-field S7 object.

Examples

field <- SchemaField("order_id", "INTEGER", ordinal_position = 1L)
SchemaDatasetFacet(list(field))
DatasourceDatasetFacet("warehouse", "postgres://warehouse")

OpenLineage datasets

Description

Dataset identifies a dataset. InputDataset and OutputDataset add lifecycle-specific facet maps and are the only dataset types accepted by a RunEvent.

Usage

Dataset(namespace, name, facets = list())

InputDataset(namespace, name, facets = list(), input_facets = list())

OutputDataset(namespace, name, facets = list(), output_facets = list())

Arguments

Value

A Dataset, InputDataset, or OutputDataset S7 object.

Examples

Dataset("postgres://warehouse", "analytics.orders")
InputDataset("postgres://warehouse", "raw.orders")
OutputDataset("postgres://warehouse", "analytics.orders")

Input and output statistics facets

Description

Input and output statistics facets

Usage

InputStatisticsInputDatasetFacet(
  row_count = NULL,
  size = NULL,
  file_count = NULL,
  producer = OPENLINEAGE_PRODUCER
)

OutputStatisticsOutputDatasetFacet(
  row_count = NULL,
  size = NULL,
  file_count = NULL,
  producer = OPENLINEAGE_PRODUCER
)

Arguments

Value

A typed input- or output-dataset facet S7 object.

Examples

InputStatisticsInputDatasetFacet(row_count = 100L)
OutputStatisticsOutputDatasetFacet(row_count = 100L, size = 2048L)

Typed job facets

Description

Constructors for SQL, source location, and job-type metadata.

Usage

SQLJobFacet(
  query,
  dialect = NULL,
  producer = OPENLINEAGE_PRODUCER,
  deleted = NULL
)

SourceCodeLocationJobFacet(
  type,
  url,
  repo_url = NULL,
  path = NULL,
  version = NULL,
  tag = NULL,
  branch = NULL,
  pull_request_number = NULL,
  producer = OPENLINEAGE_PRODUCER,
  deleted = NULL
)

EmissionPattern(event_trigger, event_content_mode, window_duration = NULL)

JobTypeJobFacet(
  processing_type,
  integration,
  job_type = NULL,
  emission_pattern = NULL,
  producer = OPENLINEAGE_PRODUCER,
  deleted = NULL
)

Arguments

Value

A typed job facet or emission-pattern S7 object.

Examples

SQLJobFacet("SELECT 1", dialect = "ansi")
SourceCodeLocationJobFacet(
  "git",
  "https://github.com/example/project/blob/main/job.R"
)
pattern <- EmissionPattern("EVENT_BASED", "COMPLETE_SNAPSHOT")
JobTypeJobFacet("BATCH", "R", emission_pattern = pattern)

Local OpenLineage transports

Description

Offline transports for testing, local development, and disabled emission. Every emit(event) method accepts a RunEvent model and returns that event invisibly after success.

Details

AccumulatingTransport stores emitted models in its public events list. Its clear() method removes all accumulated events and returns the transport invisibly.

ConsoleTransport writes one JSON event to stream. Set pretty = TRUE for indented output.

NoopTransport validates an event but performs no output or storage.

All transports provide close(), which returns TRUE invisibly because the local transports have no pending work.

Value

An R6 transport object.

Super class

OpenLineageTransport -> AccumulatingTransport

Public fields

Methods

Public methods

• AccumulatingTransport$new()

• AccumulatingTransport$emit()

• AccumulatingTransport$clear()

• AccumulatingTransport$clone()

AccumulatingTransport$new()

Create an empty accumulating transport.

Usage

AccumulatingTransport$new()

Returns

A new AccumulatingTransport object.

AccumulatingTransport$emit()

Store an event.

Usage

AccumulatingTransport$emit(event)

Arguments

Returns

event, invisibly.

AccumulatingTransport$clear()

Remove all stored events.

Usage

AccumulatingTransport$clear()

Returns

The transport, invisibly.

AccumulatingTransport$clone()

The objects of this class are cloneable with this method.

Usage

AccumulatingTransport$clone(deep = FALSE)

Arguments

Super class

OpenLineageTransport -> ConsoleTransport

Methods

Public methods

• ConsoleTransport$new()

• ConsoleTransport$emit()

• ConsoleTransport$clone()

ConsoleTransport$new()

Create a console transport.

Usage

ConsoleTransport$new(stream = stdout(), pretty = TRUE)

Arguments

Returns

A new ConsoleTransport object.

ConsoleTransport$emit()

Serialize and write an event to the configured connection.

Usage

ConsoleTransport$emit(event)

Arguments

Returns

event, invisibly.

ConsoleTransport$clone()

The objects of this class are cloneable with this method.

Usage

ConsoleTransport$clone(deep = FALSE)

Arguments

Super class

OpenLineageTransport -> NoopTransport

Methods

Public methods

• NoopTransport$emit()

• NoopTransport$clone()

NoopTransport$emit()

Validate an event without emitting it.

Usage

NoopTransport$emit(event)

Arguments

Returns

event, invisibly.

NoopTransport$clone()

The objects of this class are cloneable with this method.

Usage

NoopTransport$clone(deep = FALSE)

Arguments

Examples

transport <- AccumulatingTransport$new()
event <- RunEvent(
  Run(new_run_id()),
  Job("example", "task"),
  event_type = "START"
)
transport$emit(event)
length(transport$events)

console <- ConsoleTransport$new(pretty = FALSE)
noop <- NoopTransport$new()
noop$emit(event)

Format an OpenLineage event time

Description

Converts an R date-time to an ISO 8601 UTC timestamp with millisecond precision, suitable for the eventTime field of an OpenLineage event.

Usage

new_event_time(time = Sys.time())

Arguments

Value

A single UTC timestamp string.

Examples

time <- as.POSIXct("2026-01-02 03:04:05", tz = "UTC")
new_event_time(time)

Generate a run identifier

Description

Generates a random UUID suitable for the runId field of an OpenLineage run. The UUID version is an implementation detail and may change.

Usage

new_run_id()

Value

A single UUID string.

Examples

new_run_id()

Create a generic OpenLineage facet

Description

Creates a facet for extensions or facet schemas without a typed constructor. Names supplied through ... are preserved exactly at the JSON boundary.

Usage

ol_facet(schema_url, ..., producer = OPENLINEAGE_PRODUCER, deleted = NULL)

Arguments

Value

A generic OpenLineage facet accepted in any facet map.

Examples

facet <- ol_facet(
  "https://example.com/CustomRunFacet.json",
  customValue = "example"
)
Run(new_run_id(), facets = list(custom = facet))

Create an OpenLineage tag

Description

Create an OpenLineage tag

Usage

ol_tag(key, value, source = NULL, field = NULL)

Arguments

Value

A tag record for a typed tags facet.

Examples

ol_tag("environment", "production", source = "USER")

OpenLineage protocol constants

Description

Constants used by openlineage when constructing OpenLineage events. OPENLINEAGE_SCHEMA_URL identifies the schema document, while OPENLINEAGE_RUN_EVENT_SCHEMA_URL identifies its RunEvent definition.

Usage

OPENLINEAGE_SCHEMA_VERSION

OPENLINEAGE_SCHEMA_URL

OPENLINEAGE_RUN_EVENT_SCHEMA_URL

OPENLINEAGE_PRODUCER

Format

Character scalars.

Examples

OPENLINEAGE_SCHEMA_VERSION
OPENLINEAGE_SCHEMA_URL
OPENLINEAGE_RUN_EVENT_SCHEMA_URL
OPENLINEAGE_PRODUCER

Typed run facets

Description

Constructors for commonly used OpenLineage run facets.

Usage

NominalTimeRunFacet(
  nominal_start_time,
  nominal_end_time = NULL,
  producer = OPENLINEAGE_PRODUCER
)

ParentRunFacet(run, job, root = NULL, producer = OPENLINEAGE_PRODUCER)

ErrorMessageRunFacet(
  message,
  programming_language,
  stack_trace = NULL,
  producer = OPENLINEAGE_PRODUCER
)

ProcessingEngineRunFacet(
  version,
  name = NULL,
  openlineage_adapter_version = NULL,
  producer = OPENLINEAGE_PRODUCER
)

Arguments

Value

A typed run facet S7 object.

Examples

NominalTimeRunFacet("2026-01-02T03:00:00.000Z")
parent_run <- Run(new_run_id())
parent_job <- Job("example", "parent")
ParentRunFacet(parent_run, parent_job)
ErrorMessageRunFacet("Query failed", "R")
ProcessingEngineRunFacet("4.6.0", name = "R")

Typed tags facets

Description

Typed tags facets

Usage

TagsJobFacet(tags = list(), producer = OPENLINEAGE_PRODUCER, deleted = NULL)

TagsRunFacet(tags = list(), producer = OPENLINEAGE_PRODUCER)

TagsDatasetFacet(
  tags = list(),
  producer = OPENLINEAGE_PRODUCER,
  deleted = NULL
)

Arguments

Value

A typed tags facet S7 object.

Examples

tag <- ol_tag("environment", "production")
TagsJobFacet(list(tag))
TagsRunFacet(list(tag))
TagsDatasetFacet(list(
  ol_tag("sensitivity", "restricted", field = "customer_id")
))

Serialize an OpenLineage value to JSON

Description

Serialize an OpenLineage value to JSON

Usage

to_openlineage_json(x, pretty = FALSE)

Arguments

Value

A single JSON string.

Examples

event <- RunEvent(
  Run("019c0000-0000-7000-8000-000000000001"),
  Job("example", "task"),
  event_type = "START",
  event_time = "2026-01-02T03:04:05.000Z"
)
to_openlineage_json(event)