Module bytewax.recovery

Classes

class KafkaRecoveryConfig (brokers, topic_prefix)

Use Kafka to store recovery data.

Uses a "progress" topic and a "state" topic with a number of partitions equal to the number of workers. Will take advantage of log compaction so that topic size is proportional to state size, not epoch count.

Use a distinct topic prefix per dataflow so recovery data is not mixed.

>>> from bytewax.execution import run_main
>>> from bytewax.inputs import TestingInputConfig
>>> from bytewax.outputs import StdOutputConfig
>>> flow = Dataflow()
>>> flow.inp("inp", TestingInputConfig(range(3)))
>>> flow.capture(StdOutputConfig())
>>> recovery_config = KafkaRecoveryConfig(
...     ["localhost:9092"],
...     "sample-dataflow",
... )
>>> run_main(
...     flow,
...     recovery_config=recovery_config,
... )  # doctest: +ELLIPSIS
(...)

If there's no previous recovery data, topics will automatically be created with the correct number of partitions and log compaction enabled

Args

brokers : List[str]

List of host:port strings of Kafka brokers.

topic_prefix : str

Prefix used for naming topics. Must be distinct per-dataflow. Two topics will be created using this prefix "topic_prefix-progress" and "topic_prefix-state".

Returns

Config object. Pass this as the recovery_config argument to your execution entry point.

Ancestors

• RecoveryConfig

Instance variables

var brokers

Return an attribute of instance, which is of type owner.

var topic_prefix

Return an attribute of instance, which is of type owner.

class RecoveryConfig

Base class for a recovery config.

This describes how each worker in a dataflow cluster should store its recovery data.

Use a specific subclass of this that matches the kind of storage system you are going to use. See the subclasses in this module.

Subclasses

• KafkaRecoveryConfig
• SqliteRecoveryConfig

class SqliteRecoveryConfig (db_dir)

Use SQLite to store recovery data.

Creates a SQLite DB per-worker in a given directory. Multiple DBs are used to allow workers to write without contention.

Use a distinct directory per dataflow so recovery data is not mixed.

>>> from bytewax.execution import run_main
>>> from bytewax.inputs import TestingInputConfig
>>> from bytewax.outputs import StdOutputConfig
>>> flow = Dataflow()
>>> flow.input("inp", TestingInputConfig(range(3)))
>>> flow.capture(StdOutputConfig())
>>> tmp_dir = TemporaryDirectory()  # We'll store this somewhere temporary for this test.
>>> recovery_config = SqliteRecoveryConfig(tmp_dir)
>>> run_main(
...     flow,
...     recovery_config=recovery_config,
... )  # doctest: +ELLIPSIS
(...)

DB files and tables will automatically be created if there's no previous recovery data.

Args

db_dir : Path

Existing directory to store per-worker DBs in. Must be distinct per-dataflow. DB files will have names like "worker0.sqlite3". You can use "." for the current directory.

Returns

Config object. Pass this as the recovery_config argument to your execution entry point.

Ancestors

• RecoveryConfig

Instance variables

var db_dir

Return an attribute of instance, which is of type owner.