Modern Data Architecture: Strategies and Best Practices

# Example Kafka topic configuration with detailed parameters
kafka-topics --create --bootstrap-server localhost:9092 \
  --replication-factor 3 \                          # Store 3 copies of each message for fault tolerance
  --partitions 8 \                                  # Split the topic into 8 partitions for parallel processing
  --topic market-data \                             # Name of the topic to store market data events
  --config retention.ms=86400000 \                  # Keep data for 24 hours (in milliseconds)
  --config min.insync.replicas=2                    # Require at least 2 replicas to be in sync for writes

// Example Flink stateful processing
// First, create a stream from Kafka consumer source
DataStream<MarketEvent> marketEvents = env.addSource(kafkaConsumer);

// Now build the processing pipeline
marketEvents
    .keyBy(event -> event.getSymbol())              // Group events by trading symbol (e.g., BTC-USD)
    .window(TumblingEventTimeWindows.of(Time.seconds(5)))  // Process in 5-second windows based on event time
    .process(new MarketAggregationFunction())       // Apply custom aggregation logic to each window
    .addSink(new AlertingSink());                   // Send results to alerting system for monitoring

# PySpark with Iceberg example - Setting up a Spark session with Iceberg integration
spark = SparkSession.builder \
    .appName("Batch Processing") \
    .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \  # Enable Iceberg extensions
    .config("spark.sql.catalog.iceberg", "org.apache.iceberg.spark.SparkCatalog") \  # Register Iceberg catalog
    .config("spark.sql.catalog.iceberg.type", "hadoop") \  # Use Hadoop catalog type
    .config("spark.sql.catalog.iceberg.warehouse", "s3://data-warehouse/iceberg") \  # S3 location for tables
    .getOrCreate()

# Time travel query - One of Iceberg's powerful features
# This query retrieves cryptocurrency market data as it existed on April 1, 2025
df = spark.sql("""
SELECT * FROM iceberg.trading.market_data 
TIMESTAMP AS OF '2025-04-01 00:00:00'  -- Point-in-time query capability
WHERE symbol = 'BTC-USD'                -- Filter for Bitcoin-USD trading pair
""")

// Kafka Streams topology - Building a data processing pipeline for order validation and routing
StreamsBuilder builder = new StreamsBuilder();      // Initialize the streams builder (core Kafka Streams object)
KStream<String, Order> orders = builder.stream("new-orders");  // Create a stream from the "new-orders" Kafka topic

// Create a new stream of validated orders by applying transformations
KStream<String, ValidatedOrder> validatedOrders = orders
    .filter((key, order) -> order.getAmount() > 0)  // Filter out orders with zero or negative amounts
    .mapValues(order -> validateOrder(order));      // Transform each order through a validation function

// Send the validated orders to a new Kafka topic
validatedOrders.to("validated-orders");             // Write results to the "validated-orders" topic

// Split the stream into multiple streams based on order type (market, limit, other)
KStream<String, ValidatedOrder>[] branches = validatedOrders.branch(
    (key, order) -> order.getType().equals("MARKET"),  // First branch: Market orders
    (key, order) -> order.getType().equals("LIMIT"),   // Second branch: Limit orders
    (key, order) -> true                               // Third branch: All other order types
);

// Send each branch to its own dedicated Kafka topic for specialized processing
branches[0].to("market-orders");                    // Market orders go to dedicated topic
branches[1].to("limit-orders");                     // Limit orders go to dedicated topic
branches[2].to("other-orders");                     // Other order types go to catch-all topic

-- Example Iceberg DDL - Creating a trading history table with optimized organization

-- Creating a table with partitioning and sorting for efficient querying
CREATE TABLE analytics.trade_history (
  trade_id BIGINT,                -- Unique identifier for each trade
  symbol STRING,                  -- Trading pair (e.g., BTC-USD, ETH-USD)
  price DECIMAL(18,8),            -- Trade price with 8 decimal precision for cryptocurrency values
  quantity DECIMAL(18,8),         -- Trade quantity with high precision
  trade_time TIMESTAMP,           -- When the trade occurred
  exchange_id STRING,             -- Source exchange (e.g., Coinbase, Binance)
  trade_type STRING               -- Type of trade (e.g., market, limit)
)
USING iceberg                     -- Specifies we're using Apache Iceberg table format
PARTITIONED BY (days(trade_time), exchange_id)  -- Partition strategy for efficient filtering
TBLPROPERTIES (
  'write.format.default' = 'parquet',  -- Store data in columnar Parquet format
  'write.parquet.compression-codec' = 'zstd',  -- Use ZSTD compression for better ratio/performance
  'sort-order' = 'symbol,trade_time DESC'  -- Organize files for faster querying by symbol and time
);

# Great Expectations data validation example - A framework for validating, documenting, and profiling data
import great_expectations as ge

# Create a validator for your dataset (works with pandas DataFrames, Spark DataFrames, SQL, etc.)
validator = ge.dataset.PandasDataset(df)

# Define and validate expectations - these are assertions about what your data should look like
results = validator.expect_column_values_to_not_be_null("trade_id")  # Every trade must have an ID
validator.expect_column_values_to_be_between("price", 0, 1000000)    # Price must be in a reasonable range
validator.expect_column_values_to_match_regex("symbol", r'^[A-Z]+-[A-Z]+$')  # Symbol format validation

# Save validation results - this documents the data quality and can be used for monitoring
validator.save_expectation_suite(discard_failed_expectations=False)  # Keep track of failed validations

# Prometheus alert example - Configuration for monitoring data pipeline freshness
groups:
- name: DataPipelineAlerts                          # Logical grouping of related alerts
  rules:
  - alert: DataFreshnessCritical                    # Alert name - describes the issue
    expr: time() - max(data_last_update_timestamp) by (dataset) > 3600  # Alert if data is >1 hour old
    for: 15m                                        # Only trigger if condition persists for 15 minutes
    labels:
      severity: critical                            # Severity level for routing and prioritization
    annotations:
      summary: "Data freshness critical for {{ $labels.dataset }}"  # Alert summary with dataset name
      description: "Dataset {{ $labels.dataset }} has not been updated in over 1 hour"  # Detailed description

-- Example Apache Ranger policy (SQL representation) - Implementing fine-grained access control
-- Apache Ranger helps implement role-based security policies across the data platform

-- Grant read-only access to data analysts, but only for specific exchanges
GRANT SELECT ON TABLE analytics.trade_history 
TO ROLE data_analysts
WHERE exchange_id IN ('Coinbase', 'Binance');  -- Row-level security filtering

-- Grant broader access to data engineers who need to modify the data
GRANT SELECT, INSERT, UPDATE ON TABLE analytics.trade_history 
TO ROLE data_engineers;  -- Full table access for the engineering team

# Example Airflow DAG with modern patterns - Orchestrating a data pipeline
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator
from airflow.sensors.external_task import ExternalTaskSensor
from airflow.utils.task_group import TaskGroup
from datetime import datetime, timedelta

# Define default parameters for all tasks in the DAG
default_args = {
    'owner': 'data_engineering',             # Team responsible for the pipeline
    'retries': 1,                            # Number of retry attempts on failure
    'retry_delay': timedelta(minutes=5),     # Wait time between retries
    'execution_timeout': timedelta(hours=2)  # Maximum allowed execution time
}

# Create the DAG definition
with DAG(
    'market_data_processing',                        # Unique identifier for this DAG
    default_args=default_args,                       # Apply the default arguments
    description='Process market data using modern patterns',
    schedule_interval='0 */3 * * *',                 # Run every 3 hours (cron format)
    start_date=datetime(2025, 4, 1),                 # When this DAG should start
    catchup=False,                                   # Don't run for historical dates
    tags=['market-data', 'production'],              # Labels for organization
) as dag:
    
    # First task: Run data quality checks before processing
    data_quality_check = PythonOperator(
        task_id='validate_source_data',              # Unique ID for this task
        python_callable=run_data_quality_checks,     # Function to execute
        op_kwargs={'dataset': 'market_data'},        # Arguments for the function
    )
    
    # Group related tasks for parallel processing of different exchanges
    with TaskGroup('process_exchanges') as process_exchanges:
        # Dynamically create a task for each exchange
        for exchange in ['binance', 'coinbase', 'kraken']:
            process_exchange = SparkSubmitOperator(
                task_id=f'process_{exchange}',        # Generate unique ID for each exchange
                application='/jobs/process_exchange.py',  # Spark job to submit
                conf={'spark.dynamicAllocation.enabled': 'true'},  # Spark config
                application_args=[exchange, '{{ ds }}'],  # Pass exchange name and execution date
                packages='org.apache.iceberg:iceberg-spark-runtime-3.3_2.12:1.1.0',  # Required dependencies
            )
    
    # Final task: Aggregate data from all exchanges
    aggregate_data = SparkSubmitOperator(
        task_id='aggregate_market_data',
        application='/jobs/aggregate_market_data.py',
        conf={'spark.sql.shuffle.partitions': '200'},  # Optimize shuffle performance
        application_args=['{{ ds }}'],                 # Pass execution date to job
    )
    
    # Define the execution order using operators
    data_quality_check >> process_exchanges >> aggregate_data  # Linear workflow

# Load images from an S3 bucket with Daft
import daft

# Create a dataframe from image files stored in S3
df = daft.from_glob_path("s3://data-bucket/images/*")

# Download and decode images in a distributed manner
df = df.with_column("image", df["path"].url.download().image.decode())

# Process the images (e.g., resize to thumbnails)
df = df.with_column("thumbnail", df["image"].image.resize(32, 32))

# Execute the pipeline
result = df.collect()