This tutorial will show you how to create a unified Apache Beam DirectRunner allows you to create a seamless pipeline in which batch or stream mode can be used. We generate synthetic, event-time–aware data and apply fixed windowing with triggers and allowed lateness to demonstrate how Apache Beam consistently handles both on-time and late events. The core logic of the aggregation is unchanged by switching the input source. This allows us to understand Beam’s event-time models, windows and panes without depending on external streaming. See the FULL CODES here.
Install -U!pip "grpcio>=1.71.2" "grpcio-status>=1.71.2"
Install apache beam crcmod with!pip.
You can import apache_beam into your beam.
from apache_beam.options.pipeline_options import PipelineOptions, StandardOptions
from apache_beam.transforms.window import FixedWindows
from apache_beam.transforms.trigger import AfterWatermark, AfterProcessingTime, AccumulationMode
from apache_beam.testing.test_stream import TestStream
Import json
Import datetime and timezone from datetimeApache Beam is installed and compatible with the latest version. Importing the Beam core APIs as well as windowing, testStream, and triggers is necessary later on in the pipeline. The Python standard modules are also used for JSON and time formatting. You can check out the FULL CODES here.
MODE =Return "stream"
WINDOW_SIZE_SECS = 60
ALLOWED_LATENESS_SECS = 120
def make_event(user_id, event_type, amount, event_time_epoch_s):
return {"user_id": user_id, "event_type": event_type, "amount": float(amount), "event_time": int(event_time_epoch_s)}
base = datetime.now(timezone.utc).replace(microsecond=0)
t0 = int(base.timestamp())
BATCH_EVENTS = [
make_event("u1", "purchase", 20, t0 + 5),
make_event("u1", "purchase", 15, t0 + 20),
make_event("u2", "purchase", 8, t0 + 35),
make_event("u1", "refund", -5, t0 + 62),
make_event("u2", "purchase", 12, t0 + 70),
make_event("u3", "purchase", 9, t0 + 75),
make_event("u2", "purchase", 3, t0 + 50),
]
The global configuration controls the window size, execution mode, and lateness. To make the windowing behaviour deterministic, we create events that have explicit timestamps for event-time. To observe Beam event-time semantics, we prepare a dataset with out-of order and late events. Click here to see the FULL CODES here.
def format_joined_record(kv):
User_id, D = kvReturn
return {
"user_id": user_id,
"count": int(d["count"][0]) if d["count"] Other than 0,
"sum_amount": float(d["sum_amount"][0]) if d["sum_amount"] Other 0.0
}
class WindowedUserAgg(beam.PTransform):
def expand(self, pcoll):
stamped = pcoll | beam.Map(lambda e: beam.window.TimestampedValue(e"["event_time"]))
windowed = stamped | beam.WindowInto(
FixedWindows(WINDOW_SIZE_SECS),
allowed_lateness=ALLOWED_LATENESS_SECS,
trigger=AfterWatermark(
early=AfterProcessingTime(10),
late=AfterProcessingTime(10),
),
accumulation_mode=AccumulationMode.ACCUMULATING,
)
keyed = windowed | beam.Map(lambda e: (e["user_id"], e["amount"]))
counts = keyed | beam.combiners.Count.PerKey()
sums = keyed | beam.CombinePerKey(sum)
Return (
{"count": counts, "sum_amount": sums}
| beam.CoGroupByKey()
| beam.Map(format_joined_record)
)This is a Beam PTransform which encapsulates the entire windowed aggregate logic. Applying fixed windows, accumulation rules and triggers to events, we then aggregate them by user, compute the counts, and totals. This transform is independent of data sources, and the logic works for both streaming and batch input. See the FULL CODES here.
class AddWindowInfo(beam.DoFn):
def process(self, element, window=beam.DoFn.WindowParam, pane_info=beam.DoFn.PaneInfoParam):
Window.start = ws
We = float (window.end)."
yield {
**element,
"window_start_utc": datetime.fromtimestamp(ws, tz=timezone.utc).strftime("%H:%M:%S"),
"window_end_utc": datetime.fromtimestamp(we, tz=timezone.utc).strftime("%H:%M:%S"),
"pane_timing": str(pane_info.timing),
"pane_is_first": pane_info.is_first,
"pane_is_last": pane_info.is_last,
}
Def build_test_stream():
Return (
TestStream()
.advance_watermark_to(t0)
.add_elements([
beam.window.TimestampedValue(make_event("u1", "purchase", 20, t0 + 5), t0 + 5),
beam.window.TimestampedValue(make_event("u1", "purchase", 15, t0 + 20), t0 + 20),
beam.window.TimestampedValue(make_event("u2", "purchase", 8, t0 + 35), t0 + 35),
])
.advance_processing_time(5)
.advance_watermark_to(t0 + 61)
.add_elements([
beam.window.TimestampedValue(make_event("u1", "refund", -5, t0 + 62), t0 + 62),
beam.window.TimestampedValue(make_event("u2", "purchase", 12, t0 + 70), t0 + 70),
beam.window.TimestampedValue(make_event("u3", "purchase", 9, t0 + 75), t0 + 75),
])
.advance_processing_time(5)
.add_elements([
beam.window.TimestampedValue(make_event("u2", "purchase", 3, t0 + 50), t0 + 50),
])
.advance_watermark_to(t0 + 121)
.advance_watermark_to_infinity()
)Each aggregated record is enhanced with window metadata and pane meta data so that we are clear on when and why the results were emitted. Beam’s timestamps can be converted to human-readable times using UTC. TestStream simulates the real-world streaming behaviour using late data, watermarks and processing time advances. Visit the FULL CODES here.
def run_batch():
with beam.Pipeline(options=PipelineOptions([]"" as p.
(
The p
WindowedUserAgg | beam.Create(BATCH_EVENTS)
| WindowedUserAgg()
| beam.ParDo(AddWindowInfo())
| beam.Map(json.dumps)
| beam.Map(print)
)
def run_stream():
Opts = PipelineOptions[])
opts.view_as(StandardOptions).streaming = True
with beam.Pipeline(options=opts) as p:
(
The p
WindowedUserAgg | build_test_stream()
| WindowedUserAgg()
| beam.ParDo(AddWindowInfo())
| beam.Map(json.dumps)
| beam.Map(print)
)
run_stream() If MODE == "stream" else run_batch()Wire everything into batch- and streamlike pipelines. Toggle between the two modes, change a flag and reuse the aggregation transformation. The output can be inspected by printing the windows directly after running the pipeline.
The Beam pipeline was able to process unbounded data streams and batch data with the same semantics. Watermarks and triggers influence the time results are released and late data can update previously calculated windows. We also focused on Beam’s conceptual foundations, which provided a solid basis for scaling later the same design in real streaming runners or production environments.
Click here to find out more FULL CODES here. Also, feel free to follow us on Twitter Don’t forget about our 100k+ ML SubReddit Subscribe Now our Newsletter. Wait! What? now you can join us on telegram as well.
Our latest releases of ai2025.devThe platform is based on the 2025 vision and turns benchmarks and activity in ecosystems into structured data that you can compare, filter and export.
