Apache Beam流处理疑问：无流类如何处理流？求架构图与示例

Great question! Let me break this down for you, since Beam's "unified batch + stream" model is one of its most powerful (and sometimes confusing) features.

Why You Don't See "Stream-Specific" Classes #

First off, the reason you aren't finding dedicated stream processing classes is intentional: Beam uses a unified API that treats batch and stream data the same way at the code level. There's no StreamPCollection or StreamTransform—instead, the difference between batch and stream is defined by whether your PCollection is bounded (finite, like a file) or unbounded (infinite, like a continuous stream of events from Kafka or Pub/Sub).

Beam abstracts away the batch/stream details into its execution "Runners" (like Google Cloud Dataflow, Apache Flink, or Apache Spark Streaming). Your code stays identical; the Runner handles adapting it to batch or stream execution under the hood.

How Beam Handles Stream Data #

At its core, stream processing in Beam just means working with an unbounded PCollection. Here's the step-by-step breakdown of how to use it:

• Use an Unbounded Data Source
  To ingest stream data, you'll use Beam's IO connectors for streaming sources, like:

  • PubsubIO (for Google Cloud Pub/Sub)
  • KafkaIO (for Apache Kafka)
  • RabbitMQIO (for RabbitMQ)
    These connectors produce unbounded PCollections out of the box.

• Write Pipeline Logic (Same as Batch!)
  All the core Beam transforms—ParDo, Map, Filter, Combine, etc.—work exactly the same way for bounded and unbounded PCollections. For example, you can parse incoming stream events, filter invalid ones, and aggregate metrics without changing your code from a batch job.

• Configure for Stream Execution
  When running your pipeline, you'll specify a Runner that supports stream processing (like Dataflow or Flink) and set any stream-specific options (like windowing policies, which are critical for aggregating infinite data).

Example Stream Processing Code (Python) #

Here's a simple example that reads from a Pub/Sub topic, processes messages, and writes to another topic:

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

def parse_message(message):
    # Parse raw stream message into a structured format
    return {"id": message.split(",")[0], "value": float(message.split(",")[1])}

def filter_valid(data):
    # Filter out invalid values
    return data["value"] > 0

if __name__ == "__main__":
    options = PipelineOptions([
        "--runner=DataflowRunner",
        "--project=your-gcp-project",
        "--region=us-central1",
        "--streaming"  # This flag tells the Runner to use stream mode
    ])
    
    with beam.Pipeline(options=options) as p:
        (
            p
            | "Read from Pub/Sub" >> beam.io.ReadFromPubSub(topic="projects/your-gcp-project/topics/input-topic")
            | "Decode bytes" >> beam.Map(lambda x: x.decode("utf-8"))
            | "Parse message" >> beam.Map(parse_message)
            | "Filter valid data" >> beam.Filter(filter_valid)
            | "Format output" >> beam.Map(lambda x: f"Valid: {x['id']} -> {x['value']}")
            | "Encode to bytes" >> beam.Map(lambda x: x.encode("utf-8"))
            | "Write to Pub/Sub" >> beam.io.WriteToPubSub(topic="projects/your-gcp-project/topics/output-topic")
        )

Notice that the only stream-specific parts are the --streaming flag and using ReadFromPubSub—the rest of the logic is identical to a batch pipeline.

Beam's Stream Processing Architecture #

Beam's layered architecture enables this unified model:

• User Pipeline: Your code, built with Beam's core API (no batch/stream distinctions)
• SDK Layer: Provides abstractions like PCollection (bounded/unbounded) and PTransform (processing logic)
• Runner Layer: Translates your pipeline into execution-specific logic. For streams, Runners handle critical tasks like:
  • Continuous ingestion of data
  • Windowing (grouping infinite data into logical batches for aggregation)
  • Handling late-arriving data
• Infrastructure Layer: The underlying compute (cloud clusters, local servers) where the Runner runs the job

While I can't embed the official architecture diagram here, think of it as a stack where your code sits at the top, and the Runner handles all the stream-specific heavy lifting below.

Key Notes for Stream Processing #

• Windowing: For aggregations on unbounded data, you'll need to use windowing (e.g., fixed windows, sliding windows) to define how to group events. Beam provides built-in window transforms like beam.WindowInto(beam.window.FixedWindows(60)).
• Watermarks: Beam uses watermarks to track the progress of stream data and handle late events—this is managed by the Runner, but you can configure behavior in your code.

Hope this clarifies how Beam handles stream data! Let me know if you want to dive deeper into any specific part.

内容的提问来源于stack exchange，提问作者Stella