hsfs.core.online_ingestion #

OnlineIngestion #

Metadata object used to provide Online Ingestion information for a feature group.

This class encapsulates the state and results of an online ingestion operation, including progress tracking and log retrieval.

feature_group `property` #

feature_group: fg_mod.FeatureGroup

Get the feature group associated with this ingestion.

id `property` #

id: int | None

Get the unique identifier for the ingestion operation.

num_entries `property` `writable` #

num_entries: int | None

Get the total number of entries to ingest.

results `property` #

results: list[online_ingestion_result.OnlineIngestionResult]

Get the list of ingestion results.

get_failures #

get_failures(
    size: int = 100,
) -> list[online_ingestion_failure.OnlineIngestionFailure]

Get the records of this ingestion that never reached the online feature store.

results reports how many rows failed; this reports which ones and why. Each failure identifies a Kafka record by topic, partition and offset, and carries the reason it was rejected - a value too long for its online column, a payload that does not match the schema, or a row the database refused.

Use it after an insert to find the rows that need correcting:

ingestion = feature_group.get_latest_online_ingestion()
ingestion.wait_for_completion()
for failure in ingestion.get_failures():
    print(failure.record_key, failure.failure_type, failure.failure_reason)

Failures are read from logs, not from a durable store

Failures are recovered from the online ingestion service's logs, so they are subject to the log retention of the cluster and are not guaranteed to be available indefinitely. A record is reported here as long as its feature group and ingestion headers could be read, even when the rest of the record could not be parsed. A record whose headers are themselves unreadable cannot be attributed to this ingestion at all, and is only visible in the raw online ingestion service logs.

PARAMETER	DESCRIPTION
`size`	Maximum number of failures to retrieve. TYPE: `int` DEFAULT: `100`

RETURNS	DESCRIPTION
`list[online_ingestion_failure.OnlineIngestionFailure]`	The failed records, or an empty list if none of this ingestion's records failed.

print_logs #

print_logs(priority: str = 'error', size: int = 20) -> None

Print logs related to the online ingestion operation from OpenSearch.

PARAMETER	DESCRIPTION
`priority`	Log priority to filter by. TYPE: `str` DEFAULT: `'error'`
`size`	Maximum number of log entries to retrieve. TYPE: `int` DEFAULT: `20`

refresh #

refresh()

Refresh the state of this OnlineIngestion object from the backend.

wait_for_completion #

wait_for_completion(options: dict[str, Any] = None)

Wait for the online ingestion operation to complete, displaying a progress bar.

PARAMETER	DESCRIPTION
`options`	Options for waiting. - "timeout" (int): Maximum time to wait in seconds (default: 60). - "period" (int): Polling period in seconds (default: 1). TYPE: `dict[str, Any]` DEFAULT: `None`

RAISES	DESCRIPTION
`Warning`	If the timeout is exceeded before completion.