Caching & Performance Best Practices¶

# prefect_config.py
from prefect import flow, task
from prefect.blocks.system import Secret
from prefect.filesystems import S3
from prefect.results import PersistedResult

# Configure S3 result storage
s3_block = S3.load("s3-results")

@task(cache_policy=CachePolicy.PERSISTENT)
def process_telemetry_data(data: dict) -> dict:
    """Process telemetry data with result persistence."""
    # Expensive computation
    result = {
        "processed_records": len(data),
        "avg_temperature": sum(data.values()) / len(data),
        "timestamp": datetime.utcnow().isoformat()
    }

    # Store result in S3
    return PersistedResult(value=result, storage=s3_block)

@flow
def telemetry_processing_flow():
    """Main telemetry processing flow."""
    # Load data
    data = load_telemetry_data()

    # Process with caching
    result = process_telemetry_data(data)

    # Store result
    store_result(result)

    return result

# Configure result storage
from prefect import get_client
client = get_client()
client.set_storage(s3_block)

# dask_config.py
import dask
from dask.distributed import Client, LocalCluster
from dask.cache import Cache

# Configure Dask caching
dask.config.set({
    'distributed.worker.memory.target': 0.8,
    'distributed.worker.memory.spill': 0.9,
    'distributed.worker.memory.pause': 0.95,
    'distributed.worker.memory.terminate': 0.98
})

# Create cluster with memory management
cluster = LocalCluster(
    n_workers=4,
    threads_per_worker=2,
    memory_limit='8GB',
    processes=True
)

client = Client(cluster)

# Enable worker memory caching
cache = Cache(2e9)  # 2GB cache
cache.register()

# Example: Cache intermediate results
@dask.delayed
def expensive_computation(data):
    """Expensive computation that should be cached."""
    # This will be cached in worker memory
    return process_data(data)

@dask.delayed
def another_computation(cached_result):
    """Uses cached result from previous computation."""
    return transform_data(cached_result)

# Use cached results
data = load_large_dataset()
result1 = expensive_computation(data)
result2 = another_computation(result1)
result3 = another_computation(result1)  # Uses cache

# Compute with caching
final_result = dask.compute(result2, result3)

# fastapi_cache.py
from fastapi import FastAPI, Depends, HTTPException
from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
from fastapi_cache.decorator import cache
import redis
import json
from typing import Optional

app = FastAPI()

# Configure Redis backend
redis_client = redis.Redis(host='redis', port=6379, db=0)
FastAPICache.init(RedisBackend(redis_client), prefix="api-cache")

@app.get("/telemetry/{sensor_id}")
@cache(expire=300)  # 5 minutes
async def get_telemetry(sensor_id: str, start_date: str, end_date: str):
    """Get telemetry data with caching."""
    # Expensive database query
    data = await query_telemetry_data(sensor_id, start_date, end_date)
    return data

@app.get("/analytics/summary")
@cache(expire=600)  # 10 minutes
async def get_analytics_summary():
    """Get analytics summary with caching."""
    # Expensive analytics computation
    summary = await compute_analytics_summary()
    return summary

@app.post("/telemetry")
async def create_telemetry(data: dict):
    """Create telemetry data and invalidate cache."""
    # Create data
    result = await create_telemetry_record(data)

    # Invalidate related caches
    await invalidate_telemetry_cache(data['sensor_id'])

    return result

async def invalidate_telemetry_cache(sensor_id: str):
    """Invalidate cache for specific sensor."""
    # Invalidate sensor-specific cache
    cache_key = f"telemetry:{sensor_id}:*"
    await redis_client.delete(cache_key)

# http_cache.py
from fastapi import FastAPI, Response
from fastapi.responses import JSONResponse
from datetime import datetime, timedelta
import hashlib

app = FastAPI()

@app.get("/tiles/{z}/{x}/{y}.mvt")
async def get_vector_tile(z: int, x: int, y: int, response: Response):
    """Get vector tile with proper cache headers."""
    # Generate ETag based on tile content
    tile_data = await get_vector_tile_data(z, x, y)
    etag = hashlib.md5(tile_data).hexdigest()

    # Set cache headers
    response.headers["ETag"] = f'"{etag}"'
    response.headers["Cache-Control"] = "public, max-age=3600"
    response.headers["Last-Modified"] = datetime.utcnow().strftime("%a, %d %b %Y %H:%M:%S GMT")

    return Response(content=tile_data, media_type="application/vnd.mapbox-vector-tile")

@app.get("/tiles/raster/{z}/{x}/{y}.png")
async def get_raster_tile(z: int, x: int, y: int, response: Response):
    """Get raster tile with proper cache headers."""
    # Generate ETag based on tile content
    tile_data = await get_raster_tile_data(z, x, y)
    etag = hashlib.md5(tile_data).hexdigest()

    # Set cache headers
    response.headers["ETag"] = f'"{etag}"'
    response.headers["Cache-Control"] = "public, max-age=7200"
    response.headers["Last-Modified"] = datetime.utcnow().strftime("%a, %d %b %Y %H:%M:%S GMT")

    return Response(content=tile_data, media_type="image/png")

# gpu_cache.py
import torch
import os
from pathlib import Path
import hashlib
import json

class GPUCache:
    def __init__(self, cache_dir: str = "/tmp/gpu_cache"):
        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(exist_ok=True)

    def get_model_path(self, model_name: str, version: str) -> Path:
        """Get cached model path."""
        cache_key = hashlib.md5(f"{model_name}:{version}".encode()).hexdigest()
        return self.cache_dir / f"{cache_key}.pt"

    def cache_model(self, model: torch.nn.Module, model_name: str, version: str):
        """Cache model to disk."""
        model_path = self.get_model_path(model_name, version)
        torch.save(model.state_dict(), model_path)

        # Cache metadata
        metadata = {
            "model_name": model_name,
            "version": version,
            "cached_at": datetime.utcnow().isoformat(),
            "model_size": model_path.stat().st_size
        }

        with open(model_path.with_suffix('.json'), 'w') as f:
            json.dump(metadata, f)

    def load_model(self, model_class: torch.nn.Module, model_name: str, version: str):
        """Load cached model."""
        model_path = self.get_model_path(model_name, version)

        if not model_path.exists():
            return None

        model = model_class()
        model.load_state_dict(torch.load(model_path))
        return model

# Usage
cache = GPUCache()

# Cache model
model = MyModel()
cache.cache_model(model, "telemetry_anomaly", "v1.0")

# Load cached model
cached_model = cache.load_model(MyModel, "telemetry_anomaly", "v1.0")

# geospatial_cache.py
import geopandas as gpd
import numpy as np
from shapely.geometry import Point
import pickle
from pathlib import Path

class GeospatialCache:
    def __init__(self, cache_dir: str = "/tmp/geo_cache"):
        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(exist_ok=True)

    def cache_isochrones(self, center_point: Point, travel_time: int, isochrones: gpd.GeoDataFrame):
        """Cache isochrone computation."""
        cache_key = f"isochrones_{center_point.x}_{center_point.y}_{travel_time}"
        cache_path = self.cache_dir / f"{cache_key}.pkl"

        with open(cache_path, 'wb') as f:
            pickle.dump(isochrones, f)

    def load_isochrones(self, center_point: Point, travel_time: int) -> gpd.GeoDataFrame:
        """Load cached isochrones."""
        cache_key = f"isochrones_{center_point.x}_{center_point.y}_{travel_time}"
        cache_path = self.cache_dir / f"{cache_key}.pkl"

        if not cache_path.exists():
            return None

        with open(cache_path, 'rb') as f:
            return pickle.load(f)

    def cache_od_matrix(self, origins: gpd.GeoDataFrame, destinations: gpd.GeoDataFrame, 
                       od_matrix: np.ndarray):
        """Cache origin-destination matrix."""
        cache_key = f"od_matrix_{len(origins)}_{len(destinations)}"
        cache_path = self.cache_dir / f"{cache_key}.pkl"

        with open(cache_path, 'wb') as f:
            pickle.dump(od_matrix, f)

    def load_od_matrix(self, origins: gpd.GeoDataFrame, destinations: gpd.GeoDataFrame) -> np.ndarray:
        """Load cached OD matrix."""
        cache_key = f"od_matrix_{len(origins)}_{len(destinations)}"
        cache_path = self.cache_dir / f"{cache_key}.pkl"

        if not cache_path.exists():
            return None

        with open(cache_path, 'rb') as f:
            return pickle.load(f)

# Usage
geo_cache = GeospatialCache()

# Cache isochrones
center = Point(-122.4194, 37.7749)
isochrones = compute_isochrones(center, 30)  # 30 minutes
geo_cache.cache_isochrones(center, 30, isochrones)

# Load cached isochrones
cached_isochrones = geo_cache.load_isochrones(center, 30)