Implementation Blueprint

# Update system and install Python/Pip
sudo apt update && sudo apt upgrade -y
sudo apt install python3 python3-pip python3-venv -y

# Create a virtual environment for Beam
python3 -m venv beam-env
source beam-env/bin/activate

# Install Apache Beam SDK (Python)
pip install apache-beam[gcp,aws]

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

# Define pipeline options
options = PipelineOptions(
    runner='DirectRunner', # Use 'FlinkRunner' or 'DataflowRunner' for production
    project='my-gcp-project',
    temp_location='gs://my-bucket/temp',
)

with beam.Pipeline(options=options) as p:
    (
        p
        | 'ReadInput' >> beam.io.ReadFromText('input.txt')
        | 'ExtractWords' >> beam.FlatMap(lambda x: x.split())
        | 'MapToCount' >> beam.Map(lambda x: (x, 1))
        | 'CountWords' >> beam.CombinePerKey(sum)
        | 'FormatOutput' >> beam.Map(lambda word_count: f"{word_count[0]}: {word_count[1]}")
        | 'WriteOutput' >> beam.io.WriteToText('output.txt')
    )

python wordcount.py

python wordcount.py \
  --runner FlinkRunner \
  --flink_master localhost:8081 \
  --environment_type DOCKER

python wordcount.py \
  --runner DataflowRunner \
  --project your-gcp-project-id \
  --region us-central1 \
  --temp_location gs://your-gcp-bucket/temp/ \
  --staging_location gs://your-gcp-bucket/staging/

import apache_beam.transforms.window as window

    # ... streaming read from Kafka ...
    | 'Windowing' >> beam.WindowInto(window.FixedWindows(60)) # 60 second windows
    | 'GroupByKey' >> beam.GroupByKey()
    # ... further processing ...

from apache_beam.testing.test_pipeline import TestPipeline
from apache_beam.testing.util import assert_that, equal_to

def test_pipeline():
    with TestPipeline() as p:
        input_data = p | beam.Create(['hello', 'world', 'hello'])
        output = input_data | beam.Map(lambda x: (x, 1)) | beam.CombinePerKey(sum)
        
        assert_that(output, equal_to([('hello', 2), ('world', 1)]))