Data Pipeline with Pub/Sub, Dataflow, BigQuery

import argparse
import json
import logging
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from datetime import datetime

class ParseEvent(beam.DoFn):
    def process(self, element):
        try:
            event = json.loads(element.decode("utf-8"))
            event["processing_time"] = datetime.utcnow().isoformat()
            yield event
        except Exception as e:
            logging.error(f"Parse error: {e}")
            yield beam.pvalue.TaggedOutput("deadletter", element)

class ValidatePurchase(beam.DoFn):
    def process(self, element):
        required = ["event_id", "user_id", "amount", "event_timestamp"]
        if all(k in element for k in required) and element["amount"] > 0:
            yield element
        else:
            yield beam.pvalue.TaggedOutput("deadletter", element)

def run(pipeline_args=None):
    pipeline_options = PipelineOptions(pipeline_args, streaming=True)

    with beam.Pipeline(options=pipeline_options) as p:
        events = (
            p
            | "ReadFromPubSub" >> beam.io.ReadFromPubSub(
                subscription="projects/<your-project>/subscriptions/order-sub"
            )
        )

        parsed = (
            events
            | "ParseJSON" >> beam.ParDo(ParseEvent()).with_outputs(
                "deadletter", main="valid"
            )
        )

        validated = (
            parsed.valid
            | "Validate" >> beam.ParDo(ValidatePurchase()).with_outputs(
                "deadletter", main="clean"
            )
        )

        (
            validated.clean
            | "WriteToBigQuery" >> beam.io.WriteToBigQuery(
                table="<your-project>:analytics.purchases",
                schema="event_id:STRING,user_id:STRING,amount:FLOAT,"
                       "item_count:INTEGER,category:STRING,"
                       "event_timestamp:TIMESTAMP,processing_time:TIMESTAMP",
                write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
                create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED
            )
        )

        deadletter = (
            {"parsed": parsed.deadletter, "validated": validated.deadletter}
            | "FlattenDeadletter" >> beam.Flatten()
            | "WriteDeadletter" >> beam.io.WriteToText(
                "gs://<your-project>-deadletter/failed-"
            )
        )

if __name__ == "__main__":
    logging.getLogger().setLevel(logging.INFO)
    run()

from google.cloud import pubsub_v1
import json
import uuid
import random
from datetime import datetime

publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path("<your-project>", "order-events")

categories = ["electronics", "clothing", "home", "books"]
for i in range(100):
    event = {
        "event_id": str(uuid.uuid4()),
        "user_id": f"user_{random.randint(1000, 9999)}",
        "amount": round(random.uniform(10, 500), 2),
        "item_count": random.randint(1, 5),
        "category": random.choice(categories),
        "event_timestamp": datetime.utcnow().isoformat()
    }
    data = json.dumps(event).encode("utf-8")
    future = publisher.publish(topic_path, data)
    print(f"Published {future.result()}")