JSONL for Data Streaming

Kinesis Producer

import boto3
import json
from datetime import datetime

kinesis = boto3.client('kinesis', region_name='us-east-1')
stream_name = 'user-events-stream'

def put_record(data, partition_key):
    """Put single record to Kinesis"""
    response = kinesis.put_record(
        StreamName=stream_name,
        Data=json.dumps(data),
        PartitionKey=partition_key
    )
    return response['SequenceNumber']

def put_records_batch(records):
    """Batch put up to 500 records"""
    kinesis_records = [
        {
            'Data': json.dumps(record['data']),
            'PartitionKey': record['partition_key']
        }
        for record in records
    ]

    response = kinesis.put_records(
        StreamName=stream_name,
        Records=kinesis_records
    )

    # Check for failures
    failed_count = response['FailedRecordCount']
    if failed_count > 0:
        print(f"Failed to put {failed_count} records")

# Send events
event = {
    'event_id': 'evt_12345',
    'user_id': 98765,
    'action': 'purchase',
    'product_id': 'prod_456',
    'amount': 99.99,
    'timestamp': datetime.utcnow().isoformat()
}

sequence_number = put_record(event, partition_key=str(event['user_id']))
print(f"Event sent with sequence number: {sequence_number}")

# Batch sending for better throughput
events = [
    {'data': {'action': 'login', 'user_id': i}, 'partition_key': str(i)}
    for i in range(100)
]
put_records_batch(events)

Kinesis Consumer

import boto3
import json
import time

kinesis = boto3.client('kinesis', region_name='us-east-1')
stream_name = 'user-events-stream'

def get_shard_iterator(shard_id, iterator_type='LATEST'):
    """Get iterator for a shard"""
    response = kinesis.get_shard_iterator(
        StreamName=stream_name,
        ShardId=shard_id,
        ShardIteratorType=iterator_type
    )
    return response['ShardIterator']

def process_records():
    """Process records from all shards"""
    # Get all shards
    response = kinesis.describe_stream(StreamName=stream_name)
    shards = response['StreamDescription']['Shards']

    for shard in shards:
        shard_id = shard['ShardId']
        shard_iterator = get_shard_iterator(shard_id)

        while True:
            response = kinesis.get_records(
                ShardIterator=shard_iterator,
                Limit=100
            )

            records = response['Records']
            for record in records:
                data = json.loads(record['Data'])
                print(f"Processing: {data}")

                # Process event
                handle_event(data)

            # Get next iterator
            shard_iterator = response['NextShardIterator']
            if not shard_iterator:
                break

            # Rate limiting
            time.sleep(1)

def handle_event(event):
    """Process individual event"""
    action = event.get('action')
    if action == 'purchase':
        process_purchase(event)
    elif action == 'refund':
        process_refund(event)

Kinesis Data Firehose

Automatically deliver streaming data to S3, Redshift, or Elasticsearch as JSONL.

import boto3
import json

firehose = boto3.client('firehose', region_name='us-east-1')
delivery_stream = 'events-to-s3'

def send_to_firehose(records):
    """Send records to Firehose for S3 delivery"""
    firehose_records = [
        {'Data': json.dumps(record) + '\n'}  # Firehose adds newlines
        for record in records
    ]

    response = firehose.put_record_batch(
        DeliveryStreamName=delivery_stream,
        Records=firehose_records
    )

    return response['FailedPutCount']

# Send events
events = [
    {'user_id': 123, 'action': 'login', 'timestamp': '2025-01-15T10:00:00Z'},
    {'user_id': 124, 'action': 'purchase', 'amount': 49.99, 'timestamp': '2025-01-15T10:01:00Z'}
]

failed = send_to_firehose(events)
print(f"Failed records: {failed}")

# Firehose Configuration (Terraform)
resource "aws_kinesis_firehose_delivery_stream" "events" {
  name        = "events-to-s3"
  destination = "s3"

  s3_configuration {
    role_arn   = aws_iam_role.firehose.arn
    bucket_arn = aws_s3_bucket.events.arn
    prefix     = "events/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/"

    # JSONL output format
    cloudwatch_logging_options {
      enabled = true
    }
  }
}

Event Store with JSONL

Store immutable events as JSONL for complete system history and state reconstruction.

# Event Store Format
{"event_id": "evt_001", "aggregate_id": "order_12345", "event_type": "OrderCreated", "timestamp": "2025-01-15T10:00:00Z", "data": {"user_id": 98765, "items": [{"product_id": "prod_1", "quantity": 2}], "total": 99.99}, "version": 1}
{"event_id": "evt_002", "aggregate_id": "order_12345", "event_type": "OrderItemAdded", "timestamp": "2025-01-15T10:01:00Z", "data": {"product_id": "prod_2", "quantity": 1, "price": 49.99}, "version": 2}
{"event_id": "evt_003", "aggregate_id": "order_12345", "event_type": "OrderConfirmed", "timestamp": "2025-01-15T10:05:00Z", "data": {"confirmation_id": "conf_abc123", "payment_method": "credit_card"}, "version": 3}
{"event_id": "evt_004", "aggregate_id": "order_12345", "event_type": "OrderShipped", "timestamp": "2025-01-15T14:00:00Z", "data": {"tracking_number": "TRACK123", "carrier": "UPS"}, "version": 4}

# Python Event Store Implementation
import json
from datetime import datetime
from typing import List, Dict

class EventStore:
    def __init__(self, filepath='events.jsonl'):
        self.filepath = filepath

    def append_event(self, aggregate_id: str, event_type: str, data: Dict, version: int):
        """Append event to store"""
        event = {
            'event_id': str(uuid.uuid4()),
            'aggregate_id': aggregate_id,
            'event_type': event_type,
            'timestamp': datetime.utcnow().isoformat(),
            'data': data,
            'version': version
        }

        with open(self.filepath, 'a') as f:
            f.write(json.dumps(event) + '\n')

        return event['event_id']

    def get_events(self, aggregate_id: str) -> List[Dict]:
        """Get all events for an aggregate"""
        events = []
        with open(self.filepath, 'r') as f:
            for line in f:
                event = json.loads(line)
                if event['aggregate_id'] == aggregate_id:
                    events.append(event)
        return events

    def rebuild_state(self, aggregate_id: str) -> Dict:
        """Reconstruct current state from events"""
        events = self.get_events(aggregate_id)
        state = {}

        for event in events:
            state = self.apply_event(state, event)

        return state

    def apply_event(self, state: Dict, event: Dict) -> Dict:
        """Apply event to state"""
        event_type = event['event_type']
        data = event['data']

        if event_type == 'OrderCreated':
            state = {
                'order_id': event['aggregate_id'],
                'user_id': data['user_id'],
                'items': data['items'],
                'total': data['total'],
                'status': 'created'
            }
        elif event_type == 'OrderItemAdded':
            state['items'].append({
                'product_id': data['product_id'],
                'quantity': data['quantity']
            })
            state['total'] += data['price']
        elif event_type == 'OrderConfirmed':
            state['status'] = 'confirmed'
            state['confirmation_id'] = data['confirmation_id']
        elif event_type == 'OrderShipped':
            state['status'] = 'shipped'
            state['tracking_number'] = data['tracking_number']

        return state

# Usage
store = EventStore()

# Append events
store.append_event('order_12345', 'OrderCreated', {'user_id': 98765, 'items': [], 'total': 0}, 1)
store.append_event('order_12345', 'OrderItemAdded', {'product_id': 'prod_1', 'quantity': 2, 'price': 99.99}, 2)
store.append_event('order_12345', 'OrderConfirmed', {'confirmation_id': 'conf_123'}, 3)

# Rebuild current state
current_state = store.rebuild_state('order_12345')
print(current_state)

Snapshots for Performance

class EventStoreWithSnapshots(EventStore):
    def __init__(self, filepath='events.jsonl', snapshot_interval=100):
        super().__init__(filepath)
        self.snapshot_file = filepath.replace('.jsonl', '_snapshots.jsonl')
        self.snapshot_interval = snapshot_interval

    def get_latest_snapshot(self, aggregate_id: str):
        """Get most recent snapshot"""
        snapshots = []
        try:
            with open(self.snapshot_file, 'r') as f:
                for line in f:
                    snapshot = json.loads(line)
                    if snapshot['aggregate_id'] == aggregate_id:
                        snapshots.append(snapshot)
        except FileNotFoundError:
            return None

        return snapshots[-1] if snapshots else None

    def save_snapshot(self, aggregate_id: str, state: Dict, version: int):
        """Save state snapshot"""
        snapshot = {
            'aggregate_id': aggregate_id,
            'state': state,
            'version': version,
            'timestamp': datetime.utcnow().isoformat()
        }

        with open(self.snapshot_file, 'a') as f:
            f.write(json.dumps(snapshot) + '\n')

    def rebuild_state(self, aggregate_id: str) -> Dict:
        """Rebuild from latest snapshot + subsequent events"""
        snapshot = self.get_latest_snapshot(aggregate_id)

        if snapshot:
            state = snapshot['state']
            from_version = snapshot['version']
        else:
            state = {}
            from_version = 0

        # Apply events after snapshot
        events = self.get_events(aggregate_id)
        events_to_apply = [e for e in events if e['version'] > from_version]

        for event in events_to_apply:
            state = self.apply_event(state, event)

        # Create new snapshot if needed
        if len(events) % self.snapshot_interval == 0:
            self.save_snapshot(aggregate_id, state, len(events))

        return state

Apache Flink

import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.json4s._
import org.json4s.jackson.JsonMethods._

object EventProcessor {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    // Read from Kafka
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "localhost:9092")
    properties.setProperty("group.id", "flink-consumer")

    val consumer = new FlinkKafkaConsumer[String](
      "user-events",
      new SimpleStringSchema(),
      properties
    )

    val stream = env.addSource(consumer)

    // Parse JSON events
    val events = stream
      .map(json => parse(json))
      .filter(event => (event \ "action").extract[String] == "purchase")

    // Window aggregation
    val revenue = events
      .keyBy(event => (event \ "user_id").extract[Int])
      .timeWindow(Time.minutes(5))
      .aggregate(new RevenueAggregator)

    // Alert on anomalies
    revenue
      .filter(_.total > 10000)
      .addSink(new AlertSink)

    env.execute("Event Processor")
  }
}

Apache Spark Structured Streaming

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *

spark = SparkSession.builder.appName("EventProcessor").getOrCreate()

# Read from Kafka as streaming DataFrame
events_df = spark \
    .readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "user-events") \
    .load()

# Parse JSON
schema = StructType([
    StructField("user_id", IntegerType()),
    StructField("action", StringType()),
    StructField("amount", DoubleType()),
    StructField("timestamp", StringType())
])

parsed_df = events_df \
    .selectExpr("CAST(value AS STRING) as json") \
    .select(from_json(col("json"), schema).alias("data")) \
    .select("data.*")

# Aggregate by window
windowed_revenue = parsed_df \
    .withWatermark("timestamp", "10 minutes") \
    .groupBy(
        window(col("timestamp"), "5 minutes"),
        col("user_id")
    ) \
    .agg(
        sum("amount").alias("total_amount"),
        count("*").alias("purchase_count")
    )

# Write to console
query = windowed_revenue \
    .writeStream \
    .outputMode("update") \
    .format("console") \
    .start()

query.awaitTermination()