Semiconductor Manufacturing Data Platform Implementation Plan

# Network Segmentation Plan
networks:
  production_floor:
    vlan: 100
    subnet: 10.1.100.0/24
    security_zone: manufacturing

  data_platform:
    vlan: 200  
    subnet: 10.1.200.0/24
    security_zone: analytics

  management:
    vlan: 300
    subnet: 10.1.300.0/24
    security_zone: management

  external:
    vlan: 400
    subnet: 10.1.400.0/24
    security_zone: dmz

-- Greenplum Master Configuration
-- /data/master/gpseg-1/postgresql.conf

max_connections = 500
shared_buffers = 32GB
work_mem = 256MB
maintenance_work_mem = 2GB
checkpoint_segments = 64
wal_buffers = 64MB
random_page_cost = 1.0
default_statistics_target = 250

-- Segment Node Optimization
-- /data/primary/gpseg0/postgresql.conf

shared_buffers = 128GB
work_mem = 512MB  
maintenance_work_mem = 4GB
effective_cache_size = 384GB
checkpoint_segments = 128
max_prepared_transactions = 500
gp_workfile_limit_files_per_query = 1000000

-- Create production database and schemas
CREATE DATABASE semiconductor_manufacturing;

CREATE SCHEMA production;
CREATE SCHEMA quality;  
CREATE SCHEMA equipment;
CREATE SCHEMA analytics;

-- Partition table for high-volume sensor data
CREATE TABLE equipment.sensor_data (
    sensor_id BIGSERIAL,
    equipment_id VARCHAR(50),
    timestamp TIMESTAMP,
    parameter_name VARCHAR(100),
    value FLOAT,
    unit VARCHAR(20),
    status VARCHAR(20)
) DISTRIBUTED BY (equipment_id)
PARTITION BY RANGE (timestamp)
(
    PARTITION p_2024_01 START ('2024-01-01') END ('2024-02-01'),
    PARTITION p_2024_02 START ('2024-02-01') END ('2024-03-01'),
    -- Continue monthly partitions
);

-- Lot tracking with hash distribution
CREATE TABLE production.lot_tracking (
    lot_id VARCHAR(50) PRIMARY KEY,
    product_id VARCHAR(50),
    wafer_count INTEGER,
    start_time TIMESTAMP,
    current_step VARCHAR(100),
    status VARCHAR(20),
    priority INTEGER
) DISTRIBUTED BY (lot_id);

# kafka/server.properties
broker.id=1
listeners=PLAINTEXT://kafka1:9092,SSL://kafka1:9093
num.network.threads=8
num.io.threads=16
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
num.partitions=12
num.recovery.threads.per.data.dir=2
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
log.retention.hours=168
log.segment.bytes=1073741824
compression.type=lz4

# Create topics for different data streams
kafka-topics.sh --create --topic equipment-sensors \
  --partitions 24 --replication-factor 3 \
  --config compression.type=lz4 \
  --config retention.ms=604800000

kafka-topics.sh --create --topic production-events \
  --partitions 12 --replication-factor 3 \
  --config compression.type=snappy

kafka-topics.sh --create --topic test-results \
  --partitions 8 --replication-factor 3 \
  --config compression.type=gzip

public class ManufacturingAnalyticsTopology {
    public static void main(String[] args) {
        TopologyBuilder builder = new TopologyBuilder();

        // Kafka spout for equipment sensors
        builder.setSpout("equipment-spout", 
            new KafkaSpout<>(getKafkaSpoutConfig("equipment-sensors")), 4);

        // Real-time aggregation bolt
        builder.setBolt("aggregation-bolt", 
            new SensorAggregationBolt(), 8)
            .shuffleGrouping("equipment-spout");

        // Anomaly detection bolt
        builder.setBolt("anomaly-bolt", 
            new AnomalyDetectionBolt(), 4)
            .fieldsGrouping("aggregation-bolt", new Fields("equipment_id"));

        // Greenplum sink bolt
        builder.setBolt("greenplum-sink", 
            new GreenplumSinkBolt(), 6)
            .shuffleGrouping("aggregation-bolt")
            .shuffleGrouping("anomaly-bolt");
    }
}

<!-- NiFi Template for MES Integration -->
<template>
    <processor>
        <name>FetchMESData</name>
        <class>org.apache.nifi.processors.standard.ExecuteSQL</class>
        <properties>
            <property>
                <name>SQL select query</name>
                <value>
                    SELECT lot_id, product_id, step_name, start_time, end_time
                    FROM mes_system.work_orders 
                    WHERE last_modified >= ?
                </value>
            </property>
        </properties>
        <scheduling>
            <period>5 min</period>
        </scheduling>
    </processor>

    <processor>
        <name>TransformMESData</name>
        <class>org.apache.nifi.processors.script.ExecuteScript</class>
        <properties>
            <property>
                <name>Script Engine</name>
                <value>python</value>
            </property>
            <property>
                <name>Script Body</name>
                <value>
                    import json
                    import time

                    flowFile = session.get()
                    if flowFile is not None:
                        # Transform MES data format
                        content = session.read(flowFile)
                        transformed_data = process_mes_data(content)

                        session.write(flowFile, transformed_data)
                        session.transfer(flowFile, REL_SUCCESS)
                </value>
            </property>
        </properties>
    </processor>
</template>

class FDCConnector:
    def __init__(self, kafka_producer, equipment_config):
        self.producer = kafka_producer
        self.config = equipment_config

    def collect_sensor_data(self):
        """Collect real-time sensor data from equipment"""
        while True:
            try:
                # Connect to equipment SECS/GEM interface
                equipment_data = self.secs_gem_client.get_sensor_readings()

                for sensor in equipment_data:
                    message = {
                        'timestamp': sensor.timestamp,
                        'equipment_id': sensor.equipment_id,
                        'parameter': sensor.parameter,
                        'value': sensor.value,
                        'alarm_status': sensor.alarm_status
                    }

                    self.producer.send('equipment-sensors', 
                                     key=sensor.equipment_id,
                                     value=json.dumps(message))

            except Exception as e:
                logger.error(f"FDC data collection error: {e}")
                time.sleep(10)

class APCController:
    def __init__(self, greenplum_conn):
        self.gp_conn = greenplum_conn

    def run_to_run_control(self, lot_id, recipe_id):
        """Implement run-to-run control logic"""
        # Query historical performance data
        sql = """
        SELECT parameter_name, AVG(actual_value) as avg_value,
               STDDEV(actual_value) as std_value
        FROM equipment.process_parameters 
        WHERE lot_id IN (
            SELECT lot_id FROM production.lot_tracking 
            WHERE product_id = (
                SELECT product_id FROM production.lot_tracking 
                WHERE lot_id = %s
            )
        )
        AND parameter_name IN ('temperature', 'pressure', 'flow_rate')
        GROUP BY parameter_name
        """

        results = self.gp_conn.execute(sql, (lot_id,))

        # Calculate adjustments based on statistical control
        adjustments = {}
        for row in results:
            if row['std_value'] > self.control_limits[row['parameter_name']]:
                adjustments[row['parameter_name']] = \
                    self.calculate_adjustment(row['avg_value'], row['std_value'])

        return adjustments

class CPTestIntegration:
    def __init__(self, kafka_producer):
        self.producer = kafka_producer

    def process_test_results(self, test_file_path):
        """Process CP test results and stream to Kafka"""
        with open(test_file_path, 'r') as f:
            test_data = self.parse_cp_file(f)

        for wafer in test_data['wafers']:
            wafer_summary = {
                'wafer_id': wafer['wafer_id'],
                'lot_id': wafer['lot_id'],
                'test_time': wafer['test_time'],
                'total_dies': wafer['total_dies'],
                'good_dies': wafer['good_dies'],
                'yield': wafer['good_dies'] / wafer['total_dies'],
                'bin_summary': wafer['bin_counts']
            }

            self.producer.send('test-results', 
                             key=wafer['wafer_id'],
                             value=json.dumps(wafer_summary))

            # Send detailed die-level data
            for die in wafer['die_data']:
                die_data = {
                    'wafer_id': wafer['wafer_id'],
                    'die_x': die['x_coord'],
                    'die_y': die['y_coord'],
                    'bin_result': die['bin'],
                    'test_parameters': die['measurements']
                }

                self.producer.send('die-level-data',
                                 key=wafer['wafer_id'],
                                 value=json.dumps(die_data))

from pyspark.sql import SparkSession
from pyspark.ml.feature import VectorAssembler
from pyspark.ml.regression import RandomForestRegressor
from pyspark.ml.pipeline import Pipeline

class YieldPredictionModel:
    def __init__(self, spark_session):
        self.spark = spark_session

    def create_training_pipeline(self):
        """Create ML pipeline for yield prediction"""

        # Feature engineering
        feature_cols = [
            'temperature_avg', 'temperature_std',
            'pressure_avg', 'pressure_std', 
            'flow_rate_avg', 'flow_rate_std',
            'cycle_time', 'equipment_utilization',
            'previous_yield', 'defect_density'
        ]

        assembler = VectorAssembler(
            inputCols=feature_cols,
            outputCol="features"
        )

        # Random Forest model
        rf = RandomForestRegressor(
            featuresCol="features",
            labelCol="yield",
            numTrees=100,
            maxDepth=10
        )

        pipeline = Pipeline(stages=[assembler, rf])
        return pipeline

    def train_model(self):
        """Train yield prediction model"""

        # Load training data from Greenplum
        training_data = self.spark.read \
            .format("jdbc") \
            .option("url", "jdbc:postgresql://greenplum-master:5432/manufacturing") \
            .option("dbtable", "analytics.yield_training_data") \
            .option("user", "analytics_user") \
            .option("password", "password") \
            .load()

        pipeline = self.create_training_pipeline()
        model = pipeline.fit(training_data)

        # Save model
        model.write().overwrite().save("hdfs://namenode:9000/models/yield_prediction")

        return model

-- Real-time equipment efficiency view
CREATE VIEW analytics.realtime_equipment_efficiency AS
SELECT 
    e.equipment_id,
    e.equipment_type,
    COUNT(CASE WHEN e.status = 'RUNNING' THEN 1 END) * 100.0 / COUNT(*) as availability,
    AVG(p.cycle_time) as avg_cycle_time,
    STDDEV(p.cycle_time) as cycle_time_variation,
    COUNT(f.failure_id) as failure_count,
    CURRENT_TIMESTAMP as last_updated
FROM equipment.equipment_status e
LEFT JOIN production.process_cycles p ON e.equipment_id = p.equipment_id
LEFT JOIN equipment.failures f ON e.equipment_id = f.equipment_id 
    AND f.failure_time >= CURRENT_TIMESTAMP - INTERVAL '24 hours'
WHERE e.last_update >= CURRENT_TIMESTAMP - INTERVAL '1 hour'
GROUP BY e.equipment_id, e.equipment_type;

-- Real-time yield trending view  
CREATE VIEW analytics.realtime_yield_trending AS
SELECT 
    p.product_id,
    DATE_TRUNC('hour', w.test_end_time) as time_bucket,
    AVG(w.yield_percentage) as hourly_yield,
    COUNT(*) as wafer_count,
    MIN(w.yield_percentage) as min_yield,
    MAX(w.yield_percentage) as max_yield,
    PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY w.yield_percentage) as median_yield
FROM quality.wafer_test_results w
JOIN production.lot_tracking l ON w.lot_id = l.lot_id
JOIN production.products p ON l.product_id = p.product_id
WHERE w.test_end_time >= CURRENT_TIMESTAMP - INTERVAL '24 hours'
GROUP BY p.product_id, DATE_TRUNC('hour', w.test_end_time)
ORDER BY time_bucket DESC;

// Tableau JavaScript API for real-time updates
var viz, workbook, activeSheet;

function initializeViz() {
    var containerDiv = document.getElementById("tableauViz");
    var url = "http://tableau-server/views/ManufacturingDashboard/Executive";

    var options = {
        width: "1200px",
        height: "800px",
        hideTabs: true,
        hideToolbar: true,
        onFirstInteractive: function() {
            workbook = viz.getWorkbook();
            activeSheet = workbook.getActiveSheet();

            // Set up auto-refresh every 30 seconds
            setInterval(refreshData, 30000);
        }
    };

    viz = new tableau.Viz(containerDiv, url, options);
}

function refreshData() {
    activeSheet.refreshAsync().then(function() {
        console.log("Dashboard refreshed at: " + new Date());
    });
}

// Initialize dashboard when page loads
document.addEventListener("DOMContentLoaded", initializeViz);

class ProductionReportGenerator:
    def __init__(self, greenplum_conn):
        self.gp_conn = greenplum_conn

    def generate_daily_report(self, report_date):
        """Generate comprehensive daily production report"""

        report_queries = {
            'production_summary': """
                SELECT 
                    product_id,
                    COUNT(DISTINCT lot_id) as lots_completed,
                    SUM(wafer_count) as total_wafers,
                    AVG(cycle_time_hours) as avg_cycle_time,
                    SUM(CASE WHEN status = 'COMPLETED' THEN wafer_count ELSE 0 END) as good_wafers
                FROM production.lot_tracking 
                WHERE DATE(end_time) = %s
                GROUP BY product_id
            """,

            'equipment_utilization': """
                SELECT 
                    equipment_type,
                    COUNT(*) as total_tools,
                    AVG(utilization_rate) as avg_utilization,
                    COUNT(CASE WHEN status = 'DOWN' THEN 1 END) as down_tools
                FROM equipment.equipment_status 
                WHERE DATE(last_update) = %s
                GROUP BY equipment_type
            """,

            'yield_performance': """
                SELECT 
                    p.product_name,
                    AVG(w.yield_percentage) as avg_yield,
                    MIN(w.yield_percentage) as min_yield,
                    MAX(w.yield_percentage) as max_yield,
                    COUNT(*) as wafers_tested
                FROM quality.wafer_test_results w
                JOIN production.lot_tracking l ON w.lot_id = l.lot_id
                JOIN production.products p ON l.product_id = p.product_id
                WHERE DATE(w.test_end_time) = %s
                GROUP BY p.product_name
            """
        }

        report_data = {}
        for section, query in report_queries.items():
            report_data[section] = self.gp_conn.execute(query, (report_date,))

        return self.format_report(report_data, report_date)

# prometheus.yml
global:
  scrape_interval: 15s
  evaluation_interval: 15s

scrape_configs:
  - job_name: 'greenplum-master'
    static_configs:
      - targets: ['greenplum-master:9100']
    metrics_path: /metrics
    scrape_interval: 10s

  - job_name: 'greenplum-segments'
    static_configs:
      - targets: 
        - 'segment-1:9100'
        - 'segment-2:9100'
        # ... all segment nodes
    scrape_interval: 15s

  - job_name: 'kafka-cluster'
    static_configs:
      - targets:
        - 'kafka-1:9308'
        - 'kafka-2:9308' 
        - 'kafka-3:9308'

rule_files:
  - "greenplum_alerts.yml"
  - "manufacturing_alerts.yml"

alerting:
  alertmanagers:
    - static_configs:
        - targets:
          - "alertmanager:9093"

-- Analyze table statistics regularly
ANALYZE production.lot_tracking;
ANALYZE equipment.sensor_data;
ANALYZE quality.wafer_test_results;

-- Create appropriate indexes
CREATE INDEX idx_lot_tracking_product_time 
ON production.lot_tracking (product_id, start_time);

CREATE INDEX idx_sensor_data_equipment_time 
ON equipment.sensor_data (equipment_id, timestamp);

-- Optimize partition pruning
SET gp_dynamic_partition_pruning = on;
SET gp_enable_fast_sri = on;
SET gp_enable_sort_limit = on;

-- Memory configuration for analytical workloads
SET gp_resqueue_memory_policy = 'eager_free';
SET work_mem = '512MB';
SET gp_workfile_limit_files_per_query = 1000000;