Google Cloud Dataflow

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

def run_pipeline():
    with beam.Pipeline(options=PipelineOptions()) as p:
        (p
         | 'Read from Pub/Sub' >> beam.io.ReadFromPubSub(subscription=subscription)
         | 'Parse JSON' >> beam.Map(json.loads)
         | 'Extract Fields' >> beam.Map(lambda x: {
             'user_id': x['user_id'],
             'event_time': x['timestamp'],
             'revenue': x.get('revenue', 0)
         })
         | 'Window into Sessions' >> beam.WindowInto(
             beam.window.Sessions(gap_size=30*60))  # 30-minute sessions
         | 'Group by User' >> beam.GroupByKey()
         | 'Calculate Metrics' >> beam.Map(calculate_session_metrics)
         | 'Write to BigQuery' >> beam.io.WriteToBigQuery(
             table='project:dataset.user_sessions',
             schema=table_schema))

run_pipeline()

Pipeline p = Pipeline.create(options);

p.apply("Read CSV Files", 
        TextIO.read().from("gs://bucket/input/*.csv"))
 .apply("Parse CSV", ParDo.of(new ParseCsvFn()))
 .apply("Validate Data", Filter.by(new ValidateDataFn()))
 .apply("Transform Records", ParDo.of(new TransformFn()))
 .apply("Group by Category", GroupByKey.create())
 .apply("Calculate Aggregates", ParDo.of(new AggregateFn()))
 .apply("Write to BigQuery", 
        BigQueryIO.writeTableRows()
            .to("project:dataset.processed_data")
            .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_TRUNCATE));

p.run().waitUntilFinish();

def create_feature_pipeline():
    return (
        p
        | 'Read Training Data' >> beam.io.ReadFromBigQuery(query=training_query)
        | 'Extract Features' >> beam.Map(extract_features)
        | 'Normalize Features' >> beam.Map(normalize_numeric_features)
        | 'Encode Categories' >> beam.Map(encode_categorical_features)
        | 'Create Feature Vectors' >> beam.Map(create_feature_vector)
        | 'Split Train/Validation' >> beam.Partition(
            lambda x, _: 0 if x['split'] == 'train' else 1, 2)
    )