diff --git a/runners/kafka-streams/build.gradle b/runners/kafka-streams/build.gradle index 616257ddd788..203168d7ec40 100644 --- a/runners/kafka-streams/build.gradle +++ b/runners/kafka-streams/build.gradle @@ -87,6 +87,16 @@ dependencies { } +// Starts the job server a portable pipeline is submitted to. Pass driver arguments with +// -PjobServerArgs="--job-port=8099,--artifact-port=8098". +tasks.register("runJobServer", JavaExec) { + group = "Application" + description = "Runs the Kafka Streams job server." + mainClass = "org.apache.beam.runners.kafka.streams.KafkaStreamsJobServerDriver" + classpath = sourceSets.main.runtimeClasspath + args = project.hasProperty("jobServerArgs") ? project.property("jobServerArgs").split(",") : [] +} + // The broker integration test drives the production runner against a real Kafka in Docker, so it // is not part of the default build. Run it with :runners:kafka-streams:brokerIntegrationTest. test { diff --git a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsRunner.java b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsRunner.java index 6a8f105bb2ee..1b530fd22ab1 100644 --- a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsRunner.java +++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsRunner.java @@ -34,6 +34,15 @@ /** * A {@link PipelineRunner} that submits portable jobs to an in-process or external Beam job service * backed by the Kafka Streams translation path. + * + *

This runner is experimental. It executes a subset of the Beam model correctly — the + * parts it supports are covered by Beam's {@code @ValidatesRunner} suite — but several capabilities + * that are core to the model are not implemented yet, among them side inputs, stateful {@code + * ParDo} and user timers, merging windows, custom {@code WindowFn}s and splittable {@code DoFn}. + * Its behaviour and its pipeline options may change. See the runner documentation for + * what is and is not supported, and #18479 for the work that remains. */ public class KafkaStreamsRunner extends PipelineRunner { diff --git a/website/www/site/content/en/documentation/runners/kafkastreams.md b/website/www/site/content/en/documentation/runners/kafkastreams.md new file mode 100644 index 000000000000..b5a14523d888 --- /dev/null +++ b/website/www/site/content/en/documentation/runners/kafkastreams.md @@ -0,0 +1,169 @@ +--- +type: runners +title: "Kafka Streams Runner" +--- + + +# Kafka Streams Runner + +The Kafka Streams Runner executes Beam pipelines on [Kafka +Streams](https://kafka.apache.org/documentation/streams/), by translating a pipeline into a Kafka +Streams topology. + +What distinguishes it from the other runners is that Kafka Streams is a library rather than a +cluster. There is no job manager and no resource manager to operate: an application is an ordinary +JVM process that reads from and writes to Kafka, and scaling it means starting more copies of that +process. Fault tolerance, state, and exactly-once processing come from Kafka itself — from consumer +groups, changelog topics, and transactions. + +That makes it worth considering if you already run Kafka and want Beam's programming model without +introducing a second distributed system to operate. + +## The runner is experimental + +**The Kafka Streams Runner is experimental.** It executes a meaningful subset of the Beam model +correctly, and the parts it does support are covered by Beam's own `@ValidatesRunner` suite, but +several capabilities that are core to the model are not implemented yet. Read [what is not +supported](#what-is-not-supported-yet) before choosing it for anything real. + +It is also aimed squarely at streaming. A pipeline over bounded data will run, but there are more +efficient choices for batch work; this runner exists for pipelines that do not end. + +## Running a pipeline + +The runner is portable: it executes user code over the Fn API, in an SDK harness, so a pipeline goes +to a job server rather than being run directly. From Java you do not have to start one yourself. + +### From Java + +Select `KafkaStreamsRunner` and point it at your Kafka cluster: + +``` +--runner=KafkaStreamsRunner \ +--bootstrapServers=localhost:9092 \ +--applicationId=my-beam-pipeline +``` + +With no `jobEndpoint` set, the runner starts a job server of its own on a dynamic port, submits to +it, and shuts it down when the pipeline finishes. Setting `--jobEndpoint` instead submits to a job +server you are already running. + +### From another SDK, or against a shared job server + +Start the job server, which listens on `localhost:8099` by default: + +``` +./gradlew :runners:kafka-streams:runJobServer +``` + +Then submit to it with the portable runner: + +``` +--runner=PortableRunner \ +--job_endpoint=localhost:8099 \ +--bootstrapServers=localhost:9092 \ +--applicationId=my-beam-pipeline +``` + +`applicationId` has no default and must be set. It becomes the Kafka Streams `application.id`, which +is the identity of the consumer group and of the runner's internal topics, so two different +pipelines sharing one would interfere with each other. + +## Pipeline options + +| Option | Default | Description | +| --- | --- | --- | +| `bootstrapServers` | `localhost:9092` | Kafka brokers the application connects to. | +| `applicationId` | *(required)* | Kafka Streams `application.id`. Must be unique per pipeline. | +| `internalParallelism` | `1` | Partitions for the internal topics the runner creates, which is the parallelism the shuffled parts of a pipeline can reach. | +| `topicReplicationFactor` | `1` | Replication factor for those topics. | +| `maxBundleSize` | `1000` | Elements per bundle, and elements taken per poll of an unbounded source. | +| `maxBundleTimeMs` | `1000` | Intended cap on how long a bundle may stay open. **Not applied yet** — see below. | +| `readCheckpointNumBundles` | `10` | Polls of an unbounded source between stores of its checkpoint mark. Larger values replay more after a restart. | +| `stateDir` | temp directory | Where Kafka Streams keeps local state. | + +### Topics the runner creates + +The runner shuffles through topics it names itself and creates before starting: a bootstrap topic +per `Impulse` and per source, and a repartition topic per `GroupByKey`. They carry a `__beam_` +prefix. Bootstrap topics always have one partition; repartition topics get `internalParallelism`, +which is what sets how many instances the parts of the pipeline behind a shuffle run across. + +Topics the pipeline itself reads or writes are never created implicitly. + +## What is supported + +* **Reading** — bounded and unbounded sources, through the primitive `Read`. +* **ParDo** — stateless, including multiple outputs. +* **GroupByKey**, and `Combine` through its GroupByKey expansion. +* **Windowing** — global, fixed and sliding windows, with the default trigger, allowed lateness, and + timestamp combiners. Windowing and triggering run through Beam's own `ReduceFnRunner`, backed by + Kafka Streams state and timers. +* **Flatten**, **Redistribute**. +* **Metrics** — user counters and distributions reported by the SDK harness surface as + `MetricResults`. +* **Exactly-once processing**, via Kafka transactions (`exactly_once_v2`). + +Because the runner is portable and reads the language-neutral pipeline proto, a pipeline built in +any Beam SDK should translate, provided it stays inside the subset above. Only the Java SDK has been +exercised so far. + +## What is not supported yet + +These are core parts of the Beam model that the runner does not implement. Each is a real gap rather +than a decision, and each is tracked: + +* **Side inputs** ([#39628](https://github.com/apache/beam/issues/39628)). +* **Stateful `ParDo` and user timers** ([#39629](https://github.com/apache/beam/issues/39629)) — + including timer families, looping timers + and processing-time timers. +* **Merging windows**, so session windows do not work, and **custom `WindowFn`s** + ([#39630](https://github.com/apache/beam/issues/39630)). The standard windows travel as URNs the + runner interprets directly; one the + user wrote themselves would have to run in the SDK harness, which is not wired up. +* **Splittable `DoFn`**, bounded or unbounded + ([#39631](https://github.com/apache/beam/issues/39631)). +* **`TestStream`** ([#39632](https://github.com/apache/beam/issues/39632)). +* **Reading a source in parallel** + ([#39626](https://github.com/apache/beam/issues/39626)). A source is split into exactly one part + and read by a single reader. A source that insists on splitting further is rejected at + translation rather than having its extra splits silently dropped. +* **A time bound on bundles** ([#39633](https://github.com/apache/beam/issues/39633)). + `maxBundleTimeMs` is accepted but has no effect: + closing a bundle from a wall-clock punctuator duplicated output against a real broker, and the + cause is not yet understood. Bundles are bounded by element count and closed on watermarks. +* **`finalizeCheckpoint`** ([#39634](https://github.com/apache/beam/issues/39634)) is not called on + an unbounded source's checkpoint mark, + so sources that rely on finalization to acknowledge data will not see it. +* **Committed metrics** ([#39635](https://github.com/apache/beam/issues/39635)) — only attempted + values are reported. + +## How it works + +A Beam pipeline arrives as a proto and is translated into a Kafka Streams `Topology`. Fused stages +of user code become processors that execute that code in an SDK harness over the Fn API; a +`GroupByKey` becomes a repartition topic plus a stateful processor; and the elements flowing between +them carry either data or a watermark report. + +Watermarks are the part with no direct Kafka Streams equivalent. Kafka Streams tracks stream-time, +which only advances when data arrives, whereas Beam needs a watermark that can advance on an idle +stream and that reflects every upstream instance. The runner therefore propagates its own watermark +reports alongside the data: a transform aggregates the reports of everything upstream of it, holds +until every partition of every upstream transform has reported, and only then lets its own watermark +advance. + +For the full design, see the [design +document](https://docs.google.com/document/d/1BBMURhSG4SxPcvvnKMTrmnKCr_jhXL6R4TBDBW7zsy8/edit) and +the tracking issue, [#18479](https://github.com/apache/beam/issues/18479). diff --git a/website/www/site/data/capability_matrix.yaml b/website/www/site/data/capability_matrix.yaml index a1afdc6f8abc..6f9b7256a2ff 100644 --- a/website/www/site/data/capability_matrix.yaml +++ b/website/www/site/data/capability_matrix.yaml @@ -26,6 +26,8 @@ capability-matrix: name: Apache Nemo - class: jet name: Hazelcast Jet + - class: kafka-streams + name: Kafka Streams - class: twister2 name: Twister2 - class: python direct @@ -80,6 +82,10 @@ capability-matrix: l1: "" l2: l3: "" + - class: kafka-streams + l1: "Yes" + l2: fully supported + l3: "Stateless ParDo runs in the SDK harness over the Fn API, including multiple outputs." - name: GroupByKey description: Grouping of key-value pairs per key, window, and pane. (See also other tabs.) values: @@ -119,6 +125,10 @@ capability-matrix: l1: "" l2: l3: "" + - class: kafka-streams + l1: "Yes" + l2: fully supported + l3: "Shuffles through a Kafka repartition topic keyed by the encoded Beam key." - name: Flatten description: Concatenates multiple homogenously typed collections together. values: @@ -158,6 +168,10 @@ capability-matrix: l1: "" l2: l3: "" + - class: kafka-streams + l1: "Yes" + l2: fully supported + l3: "" - name: Combine description: 'Application of an associative, commutative operation over all values ("globally") or over all values associated with each key ("per key"). Can be implemented using ParDo, but often more efficient implementations exist.' values: @@ -197,6 +211,10 @@ capability-matrix: l1: "" l2: l3: "" + - class: kafka-streams + l1: "Yes" + l2: fully supported + l3: "Executed through the GroupByKey expansion; there is no lifted pre-combine yet." - name: Composite Transforms description: Allows easy extensibility for library writers. In the near future, we expect there to be more information provided at this level -- customized metadata hooks for monitoring, additional runtime/environment hooks, etc. values: @@ -236,6 +254,10 @@ capability-matrix: l1: "" l2: l3: "" + - class: kafka-streams + l1: "Yes" + l2: fully supported + l3: "" - name: Side Inputs description: Side inputs are additional PCollections whose contents are computed during pipeline execution and then made accessible to DoFn code. The exact shape of the side input depends both on the PCollectionView used to describe the access pattern (interable, map, singleton) and the window of the element from the main input that is currently being processed. values: @@ -275,6 +297,10 @@ capability-matrix: l1: "" l2: l3: "" + - class: kafka-streams + l1: "No" + l2: not implemented + l3: "Stages run without a side input handler." - name: Source API description: Allows users to provide additional input sources. Supports both bounded and unbounded data. Includes hooks necessary to provide efficient parallelization (size estimation, progress information, dynamic splitting, etc). values: @@ -314,6 +340,10 @@ capability-matrix: l1: "" l2: l3: "" + - class: kafka-streams + l1: "Partially" + l2: bounded and unbounded, read by a single reader + l3: "A source is split into exactly one part; reading several splits in parallel is not supported." - name: Metrics description: Allow transforms to gather simple metrics across bundles in a PTransform. Provide a mechanism to obtain both committed and attempted metrics. Semantically similar to using an additional output, but support partial results as the transform executes, and support both committed and attempted values. Will likely want to augment Metrics to be more useful for processing unbounded data by making them windowed. values: @@ -353,6 +383,10 @@ capability-matrix: l1: "" l2: l3: "" + - class: kafka-streams + l1: "Partially" + l2: attempted metrics only + l3: "User metrics reported by the SDK harness surface as attempted values; committed values are not available." - name: Stateful Processing description: Allows fine-grained access to per-key, per-window persistent state. Necessary for certain use cases (e.g. high-volume windows which store large amounts of data, but typically only access small portions of it; complex state machines; etc.) that are not easily or efficiently addressed via Combine or GroupByKey+ParDo. values: @@ -392,6 +426,10 @@ capability-matrix: l1: "" l2: l3: "" + - class: kafka-streams + l1: "No" + l2: not implemented + l3: "User state is not wired to the harness." - description: Bounded Splittable DoFn Support Status anchor: what color-y: "fff" @@ -440,6 +478,10 @@ capability-matrix: l1: "Yes" l2: l3: + - class: kafka-streams + l1: "No" + l2: not implemented + l3: "" - name: Side Inputs description: "" values: @@ -479,6 +521,10 @@ capability-matrix: l1: l2: l3: + - class: kafka-streams + l1: "No" + l2: not implemented + l3: "" - name: Splittable DoFn Initiated Checkpointing description: "" values: @@ -518,6 +564,10 @@ capability-matrix: l1: "Yes" l2: l3: + - class: kafka-streams + l1: "No" + l2: not implemented + l3: "" - name: Dynamic Splitting description: "" values: @@ -557,6 +607,10 @@ capability-matrix: l1: "Yes" l2: Only with Python SDK l3: + - class: kafka-streams + l1: "No" + l2: not implemented + l3: "" - name: Bundle Finalization description: "" values: @@ -596,6 +650,10 @@ capability-matrix: l1: "Yes" l2: l3: + - class: kafka-streams + l1: "No" + l2: not implemented + l3: "" - description: Unbounded Splittable DoFn Support Status anchor: what color-y: "fff" @@ -644,6 +702,10 @@ capability-matrix: l1: "Yes" l2: l3: + - class: kafka-streams + l1: "No" + l2: not implemented + l3: "" - name: Side Inputs description: "" values: @@ -683,6 +745,10 @@ capability-matrix: l1: l2: l3: + - class: kafka-streams + l1: "No" + l2: not implemented + l3: "" - name: Splittable DoFn Initiated Checkpointing description: "" values: @@ -722,6 +788,10 @@ capability-matrix: l1: "Yes" l2: l3: + - class: kafka-streams + l1: "No" + l2: not implemented + l3: "" - name: Dynamic Splitting description: "" values: @@ -761,6 +831,10 @@ capability-matrix: l1: "No" l2: l3: + - class: kafka-streams + l1: "No" + l2: not implemented + l3: "" - name: Bundle Finalization description: "" values: @@ -800,6 +874,10 @@ capability-matrix: l1: "Yes" l2: l3: + - class: kafka-streams + l1: "No" + l2: not implemented + l3: "" - description: Where in event time? anchor: where color-y: "fff" @@ -848,6 +926,10 @@ capability-matrix: l1: "Yes" l2: supported l3: "" + - class: kafka-streams + l1: "Yes" + l2: fully supported + l3: "" - name: Fixed windows description: Fixed-size, timestamp-based windows. (Hourly, Daily, etc) values: @@ -887,6 +969,10 @@ capability-matrix: l1: "Yes" l2: supported l3: "" + - class: kafka-streams + l1: "Yes" + l2: fully supported + l3: "Windowing runs through Beam's ReduceFnRunner over Kafka Streams state and timers." - name: Sliding windows description: Possibly overlapping fixed-size timestamp-based windows (Every minute, use the last ten minutes of data.) values: @@ -926,6 +1012,10 @@ capability-matrix: l1: "Yes" l2: supported l3: "" + - class: kafka-streams + l1: "Yes" + l2: fully supported + l3: "" - name: Session windows description: Based on bursts of activity separated by a gap size. Different per key. values: @@ -965,6 +1055,10 @@ capability-matrix: l1: "Yes" l2: supported l3: "" + - class: kafka-streams + l1: "No" + l2: not implemented + l3: "Sessions are merging windows, which the windowing implementation does not handle yet." - name: Custom windows description: All windows must implement BoundedWindow, which specifies a max timestamp. Each WindowFn assigns elements to an associated window. values: @@ -1004,6 +1098,10 @@ capability-matrix: l1: "Yes" l2: supported l3: "" + - class: kafka-streams + l1: "No" + l2: not implemented + l3: "Only the standard WindowFns, which travel as URNs the runner interprets directly." - name: Custom merging windows description: A custom WindowFn additionally specifies whether and how to merge windows. values: @@ -1043,6 +1141,10 @@ capability-matrix: l1: "Yes" l2: supported l3: "" + - class: kafka-streams + l1: "No" + l2: not implemented + l3: "" - name: Timestamp control description: For a grouping transform, such as GBK or Combine, an OutputTimeFn specifies (1) how to combine input timestamps within a window and (2) how to merge aggregated timestamps when windows merge. values: @@ -1082,6 +1184,10 @@ capability-matrix: l1: "Yes" l2: supported l3: "" + - class: kafka-streams + l1: "Yes" + l2: fully supported + l3: "Timestamp combiners are applied by ReduceFnRunner." - description: When in processing time? anchor: when @@ -1131,6 +1237,10 @@ capability-matrix: l1: "Yes" l2: fully supported l3: "" + - class: kafka-streams + l1: "No" + l2: default trigger only + l3: "" - name: Event-time triggers description: Triggers that fire in response to event-time completeness signals, such as watermarks progressing. values: @@ -1170,6 +1280,10 @@ capability-matrix: l1: "Yes" l2: fully supported l3: "" + - class: kafka-streams + l1: "Partially" + l2: the default trigger only + l3: "Panes fire when the watermark passes the end of the window; other event-time triggers are untested." - name: Processing-time triggers description: Triggers that fire in response to processing-time advancing. @@ -1210,6 +1324,10 @@ capability-matrix: l1: "Yes" l2: fully supported l3: "" + - class: kafka-streams + l1: "No" + l2: not implemented + l3: "Processing-time timers are not wired up." - name: Count triggers description: Triggers that fire after seeing at least N elements. @@ -1250,6 +1368,10 @@ capability-matrix: l1: "Yes" l2: fully supported l3: "" + - class: kafka-streams + l1: "No" + l2: not implemented + l3: "" - name: Composite triggers description: Triggers which compose other triggers in more complex structures, such as logical AND, logical OR, early/on-time/late, etc. @@ -1290,6 +1412,10 @@ capability-matrix: l1: "Yes" l2: fully supported l3: "" + - class: kafka-streams + l1: "No" + l2: not implemented + l3: "" - name: Allowed lateness description: A way to bound the useful lifetime of a window (in event time), after which any unemitted results may be materialized, the window contents may be garbage collected, and any addtional late data that arrive for the window may be discarded. @@ -1330,6 +1456,10 @@ capability-matrix: l1: "Yes" l2: fully supported l3: "" + - class: kafka-streams + l1: "Yes" + l2: fully supported + l3: "Allowed lateness and late-data dropping are applied by ReduceFnRunner." - name: Timers description: A fine-grained mechanism for performing work at some point in the future, in either the event-time or processing-time domain. Useful for orchestrating delayed events, timeouts, etc in complex state per-key, per-window state machines. @@ -1370,6 +1500,10 @@ capability-matrix: l1: "Yes" l2: "Partially" l3: "" + - class: kafka-streams + l1: "No" + l2: not implemented + l3: "User timers are not wired to the harness." - description: How do refinements relate? anchor: how @@ -1419,6 +1553,10 @@ capability-matrix: l1: "Yes" l2: fully supported l3: "" + - class: kafka-streams + l1: "Yes" + l2: fully supported + l3: "" - name: Accumulating description: Elements are accumulated in state across multiple pane firings for the same window. @@ -1459,6 +1597,10 @@ capability-matrix: l1: "Yes" l2: fully supported l3: "" + - class: kafka-streams + l1: "Yes" + l2: fully supported + l3: "" - description: Additional common features not yet part of the Beam model anchor: misc @@ -1508,6 +1650,10 @@ capability-matrix: l1: l2: l3: + - class: kafka-streams + l1: "No" + l2: not implemented + l3: "" - name: Checkpoint description: APIs and semantics for saving a pipeline checkpoint are under discussion. This would be a runner-specific materialization of the pipeline state required to resume or duplicate the pipeline. values: @@ -1547,6 +1693,10 @@ capability-matrix: l1: l2: l3: + - class: kafka-streams + l1: "No" + l2: not implemented + l3: "" - name: Key-ordered delivery description: The runner offers guarantees for the order in which elements are passed in between operations. See per-key ordering semantics. values: @@ -1586,3 +1736,8 @@ capability-matrix: l1: "Unverified" l2: l3: + + - class: kafka-streams + l1: "No" + l2: not implemented + l3: "" \ No newline at end of file diff --git a/website/www/site/layouts/partials/section-menu/en/runners.html b/website/www/site/layouts/partials/section-menu/en/runners.html index 337debf3ecec..6119debe8781 100644 --- a/website/www/site/layouts/partials/section-menu/en/runners.html +++ b/website/www/site/layouts/partials/section-menu/en/runners.html @@ -19,4 +19,5 @@

  • Apache Spark
  • Google Cloud Dataflow
  • Hazelcast Jet
    • +
  • Kafka Streams
  • Twister2