Основы Data Science: Big Data Analytics, ML-конвейеры, ETL и визуализация

Это руководство охватывает ключевые концепции Data Science вместе с Big Data Analytics, Machine Learning конвейерами, ETL-процессами и визуализацией данных, подкреплённые практическими примерами.

Вкратце

Data Science объединяет статистику, программирование и предметное знание для извлечения ценной информации из данных. Big Data обрабатывает огромные объёмы информации, ML-конвейеры автоматизируют модели, ETL подготавливает данные к анализу, а визуализация делает результаты понятными.

Краткое определение

Data Science — это междисциплинарная область, использующая научные методы, процессы, алгоритмы и системы для извлечения знаний и ценных сведений из структурированных и неструктурированных данных.

Основные направления:

Big Data Analytics

• Суть: обработка колоссальных объёмов данных
• Модель 3V: Volume, Velocity, Variety
• Технологии: Hadoop, Spark, NoSQL базы данных
• Применение: анализ в реальном времени, предсказательная аналитика

Machine Learning конвейеры

• Суть: автоматизированная оркестрация ML-рабочих процессов
• Этапы: сбор данных → предобработка → обучение → оценка → развёртывание
• Инструменты: Scikit-learn, TensorFlow, MLflow, Airflow
• MLOps: версионирование, мониторинг, переобучение моделей

ETL-процессы

• Extract: извлечение данных из различных источников
• Transform: очистка и преобразование данных
• Load: загрузка данных в целевые системы
• Инструменты: Apache NiFi, Talend, AWS Glue

Data Visualization

• Суть: визуальное представление данных и полученных выводов
• Типы графиков: столбчатые, линейные, точечные, тепловые карты, древовидные диаграммы
• Инструменты: Matplotlib, Seaborn, Plotly, Tableau
• Принципы: наглядность, точность, эффективность

Ключевые пункты для запоминания

• Data Science: междисциплинарное направление анализа данных
• Big Data: большие, быстрые, разнообразные объёмы информации
• Machine Learning: автоматизированное обнаружение закономерностей в данных
• ETL: процесс подготовки данных (Extract-Transform-Load)
• Visualization: визуальное представление информации
• Pipelines: автоматизированные процессы обработки данных
• Analytics: статистический и исследовательский анализ
• Актуальное знание: современная аналитика данных и Business Intelligence

Основные компоненты

1. Сбор данных: источники и накопление информации
2. Хранение данных: размещение и организация
3. Обработка данных: очистка и трансформация
4. Анализ данных: статистический и исследовательский подход
5. Machine Learning: модели и алгоритмы
6. Визуализация данных: графическое представление результатов
7. Развёртывание моделей: применение в production
8. Мониторинг: отслеживание качества и обнаружение дрейфа

Практические примеры

1. Обработка больших данных с Apache Spark и Python

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *
from pyspark.ml.feature import VectorAssembler, StandardScaler
from pyspark.ml.regression import LinearRegression
from pyspark.ml.evaluation import RegressionEvaluator
import matplotlib.pyplot as plt
import seaborn as sns

# Big Data Processing Demo
class BigDataAnalytics:
    
    def __init__(self, app_name="BigDataAnalytics"):
        """Spark Session initialisieren"""
        self.spark = SparkSession.builder \
            .appName(app_name) \
            .config("spark.sql.adaptive.enabled", "true") \
            .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \
            .config("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128MB") \
            .getOrCreate()
        
        print(f"Spark Session erstellt: {self.spark.version}")
    
    def create_sample_data(self):
        """Beispieldaten für Big Data Demo erstellen"""
        import random
        from datetime import datetime, timedelta
        
        # Simulierte Verkaufsdaten
        data = []
        products = ["Laptop", "Smartphone", "Tablet", "Headphones", "Mouse", "Keyboard"]
        regions = ["North", "South", "East", "West", "Central"]
        
        start_date = datetime(2023, 1, 1)
        
        for i in range(1000000):  # 1M Datensätze
            date = start_date + timedelta(days=random.randint(0, 365))
            record = {
                "transaction_id": f"T{i:06d}",
                "date": date.strftime("%Y-%m-%d"),
                "product": random.choice(products),
                "region": random.choice(regions),
                "quantity": random.randint(1, 10),
                "unit_price": round(random.uniform(10, 1000), 2),
                "customer_id": f"C{random.randint(1, 50000):05d}",
                "sales_rep": f"SR{random.randint(1, 100):03d}"
            }
            record["total_amount"] = record["quantity"] * record["unit_price"]
            data.append(record)
        
        # Spark DataFrame erstellen
        schema = StructType([
            StructField("transaction_id", StringType(), True),
            StructField("date", StringType(), True),
            StructField("product", StringType(), True),
            StructField("region", StringType(), True),
            StructField("quantity", IntegerType(), True),
            StructField("unit_price", DoubleType(), True),
            StructField("customer_id", StringType(), True),
            StructField("sales_rep", StringType(), True),
            StructField("total_amount", DoubleType(), True)
        ])
        
        df = self.spark.createDataFrame(data, schema)
        print(f"Beispieldaten erstellt: {df.count()} Zeilen")
        
        return df
    
    def basic_analytics(self, df):
        """Grundlegende Analytics durchführen"""
        print("=== Grundlegende Analytics ===")
        
        # Schema anzeigen
        print("Schema:")
        df.printSchema()
        
        # Statistische Zusammenfassung
        print("\nStatistische Zusammenfassung:")
        df.describe().show()
        
        # Top-Produkte nach Umsatz
        print("\nTop-Produkte nach Umsatz:")
        product_sales = df.groupBy("product") \
            .agg(sum("total_amount").alias("total_sales"),
                 count("transaction_id").alias("transaction_count")) \
            .orderBy(col("total_sales").desc())
        
        product_sales.show()
        
        # Regionale Analyse
        print("\nRegionale Umsatzanalyse:")
        regional_sales = df.groupBy("region") \
            .agg(sum("total_amount").alias("total_sales"),
                 avg("total_amount").alias("avg_transaction"),
                 count("transaction_id").alias("transaction_count")) \
            .orderBy(col("total_sales").desc())
        
        regional_sales.show()
        
        # Zeitliche Analyse
        print("\nMonatliche Umsatzentwicklung:")
        monthly_sales = df.withColumn("month", substring(col("date"), 1, 7)) \
            .groupBy("month") \
            .agg(sum("total_amount").alias("monthly_sales"),
                 count("transaction_id").alias("transaction_count")) \
            .orderBy("month")
        
        monthly_sales.show()
        
        return product_sales, regional_sales, monthly_sales
    
    def advanced_analytics(self, df):
        """Fortgeschrittene Analytics mit Window Functions"""
        print("\n=== Fortgeschrittene Analytics ===")
        
        # Window Functions für laufende Summen
        from pyspark.sql.window import Window
        
        # Monatliche kumulative Umsätze
        window_spec = Window.partitionBy("month").orderBy("date") \
            .rowsBetween(Window.unboundedPreceding, Window.currentRow)
        
        monthly_cumulative = df.withColumn("month", substring(col("date"), 1, 7)) \
            .groupBy("month", "date") \
            .agg(sum("total_amount").alias("daily_sales")) \
            .withColumn("cumulative_sales", sum("daily_sales").over(window_spec)) \
            .orderBy("month", "date")
        
        print("Monatliche kumulative Umsätze:")
        monthly_cumulative.show(20)
        
        # Best-Performing Sales Reps
        rep_performance = df.groupBy("sales_rep") \
            .agg(sum("total_amount").alias("total_sales"),
                 count("transaction_id").alias("transaction_count"),
                 avg("total_amount").alias("avg_transaction")) \
            .withColumn("performance_rank", percent_rank().over(Window.orderBy(col("total_sales").desc()))) \
            .filter(col("performance_rank") <= 0.1)  # Top 10%
        
        print("\nTop 10% Sales Reps:")
        rep_performance.show()
        
        # Produkt-Korrelationen
        product_correlation = df.groupBy("customer_id", "product") \
            .agg(sum("quantity").alias("product_quantity")) \
            .groupBy("customer_id") \
            .pivot("product") \
            .agg(sum("product_quantity")) \
            .fillna(0)
        
        print("\nProdukt-Korrelationsmatrix (Top 10 Kunden):")
        product_correlation.limit(10).show()
        
        return monthly_cumulative, rep_performance, product_correlation
    
    def machine_learning_pipeline(self, df):
        """Machine Learning Pipeline für Umsatzvorhersage"""
        print("\n=== Machine Learning Pipeline ===")
        
        # Features für ML vorbereiten
        feature_data = df.groupBy("product", "region", "month") \
            .agg(sum("total_amount").alias("target_sales"),
                 count("transaction_id").alias("transaction_count"),
                 avg("unit_price").alias("avg_unit_price"),
                 sum("quantity").alias("total_quantity"))
        
        # Kategoriale Features encodieren
        from pyspark.ml.feature import StringIndexer, OneHotEncoder
        
        # Product Indexer
        product_indexer = StringIndexer(inputCol="product", outputCol="product_index")
        product_indexed = product_indexer.fit(feature_data).transform(feature_data)
        
        # Region Indexer
        region_indexer = StringIndexer(inputCol="region", outputCol="region_index")
        region_indexed = region_indexer.fit(product_indexed).transform(product_indexed)
        
        # One-Hot Encoding
        encoder = OneHotEncoder(inputCols=["product_index", "region_index"],
                               outputCols=["product_encoded", "region_encoded"])
        encoded_data = encoder.fit(region_indexed).transform(region_indexed)
        
        # Feature Vector erstellen
        assembler = VectorAssembler(
            inputCols=["transaction_count", "avg_unit_price", "total_quantity",
                       "product_encoded", "region_encoded"],
            outputCol="features"
        )
        
        assembled_data = assembler.transform(encoded_data)
        
        # Features skalieren
        scaler = StandardScaler(inputCol="features", outputCol="scaled_features")
        scaler_model = scaler.fit(assembled_data)
        scaled_data = scaler_model.transform(assembled_data)
        
        # Train-Test Split
        train_data, test_data = scaled_data.randomSplit([0.8, 0.2], seed=42)
        
        print(f"Trainingsdaten: {train_data.count()} Zeilen")
        print(f"Testdaten: {test_data.count()} Zeilen")
        
        # Lineare Regression
        lr = LinearRegression(featuresCol="scaled_features", 
                            labelCol="target_sales",
                            predictionCol="predicted_sales")
        
        lr_model = lr.fit(train_data)
        
        # Vorhersagen auf Testdaten
        predictions = lr_model.transform(test_data)
        
        # Modell evaluieren
        evaluator = RegressionEvaluator(labelCol="target_sales",
                                     predictionCol="predicted_sales",
                                     metricName="rmse")
        
        rmse = evaluator.evaluate(predictions)
        r2 = evaluator.setMetricName("r2").evaluate(predictions)
        
        print(f"Modell-Evaluation:")
        print(f"RMSE: {rmse:.2f}")
        print(f"R²: {r2:.4f}")
        
        # Feature Importance anzeigen
        print(f"\nFeature Coefficients:")
        feature_names = ["transaction_count", "avg_unit_price", "total_quantity",
                       "product_encoded", "region_encoded"]
        
        for i, (name, coef) in enumerate(zip(feature_names, lr_model.coefficients)):
            print(f"{name}: {coef:.4f}")
        
        return lr_model, predictions, scaler_model
    
    def real_time_processing_simulation(self, df):
        """Echtzeitverarbeitung simulieren"""
        print("\n=== Echtzeitverarbeitung Simulation ===")
        
        # Streaming DataFrame simulieren
        # In der Praxis würde dies von Kafka, Kinesis etc. kommen
        
        # Aggregationen für Echtzeit-Monitoring
        real_time_metrics = df.groupBy("product", "region") \
            .agg(count("transaction_id").alias("transaction_count"),
                 sum("total_amount").alias("total_sales"),
                 avg("total_amount").alias("avg_transaction")) \
            .orderBy(col("total_sales").desc())
        
        print("Echtzeit-Metriken:")
        real_time_metrics.show()
        
        # Anomalie-Erkennung (vereinfacht)
        from pyspark.sql.functions import stddev
        
        # Statistische Grenzen für Anomalie-Erkennung
        stats = df.groupBy("product") \
            .agg(avg("total_amount").alias("avg_amount"),
                 stddev("total_amount").alias("std_amount")) \
            .withColumn("upper_bound", col("avg_amount") + 2 * col("std_amount")) \
            .withColumn("lower_bound", col("avg_amount") - 2 * col("std_amount"))
        
        print("Anomalie-Grenzen pro Produkt:")
        stats.show()
        
        return real_time_metrics, stats
    
    def data_export(self, df, predictions):
        """Daten für weitere Verarbeitung exportieren"""
        print("\n=== Daten Export ===")
        
        # Ergebnisse in verschiedene Formate exportieren
        output_path = "/tmp/big_data_results"
        
        # Analytics Ergebnisse
        product_sales = df.groupBy("product") \
            .agg(sum("total_amount").alias("total_sales")) \
            .orderBy(col("total_sales").desc())
        
        product_sales.write.mode("overwrite").parquet(f"{output_path}/product_sales")
        print(f"Produkt-Umsätze exportiert nach: {output_path}/product_sales")
        
        # ML-Vorhersagen
        predictions.select("product", "region", "target_sales", "predicted_sales") \
            .write.mode("overwrite").parquet(f"{output_path}/predictions")
        
        print(f"ML-Vorhersagen exportiert nach: {output_path}/predictions")
        
        # Zusammenfassende Statistiken
        summary_stats = df.agg(
            count("transaction_id").alias("total_transactions"),
            sum("total_amount").alias("total_revenue"),
            avg("total_amount").alias("avg_transaction_value"),
            countDistinct("customer_id").alias("unique_customers"),
            countDistinct("product").alias("unique_products")
        )
        
        summary_stats.write.mode("overwrite").json(f"{output_path}/summary_stats")
        print(f"Zusammenfassende Statistiken exportiert nach: {output_path}/summary_stats")
    
    def cleanup(self):
        """Ressourcen freigeben"""
        self.spark.stop()
        print("Spark Session beendet")

# Big Data Demo ausführen
def big_data_demo():
    analytics = BigDataAnalytics()
    
    try:
        # Daten erstellen
        df = analytics.create_sample_data()
        
        # Grundlegende Analytics
        product_sales, regional_sales, monthly_sales = analytics.basic_analytics(df)
        
        # Fortgeschrittene Analytics
        monthly_cumulative, rep_performance, product_correlation = analytics.advanced_analytics(df)
        
        # Machine Learning Pipeline
        lr_model, predictions, scaler_model = analytics.machine_learning_pipeline(df)
        
        # Echtzeitverarbeitung
        real_time_metrics, stats = analytics.real_time_processing_simulation(df)
        
        # Daten exportieren
        analytics.data_export(df, predictions)
        
        print("\nBig Data Analytics Demo erfolgreich abgeschlossen!")
        
    except Exception as e:
        print(f"Fehler in Big Data Demo: {e}")
    finally:
        analytics.cleanup()

if __name__ == "__main__":
    big_data_demo()

2. ETL Pipeline с Apache Airflow

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.postgres.operators.postgres import PostgresOperator
from airflow.providers.postgres.hooks.postgres import PostgresHook
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
from airflow.providers.amazon.aws.operators.s3 import S3UploadOperator
from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator
from airflow.providers.google.cloud.hooks.bigquery import BigQueryHook
from airflow.models import Variable
from airflow.utils.dates import days_ago
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import json
import logging

# ETL Pipeline Configuration
default_args = {
    'owner': 'data-science-team',
    'depends_on_past': False,
    'start_date': days_ago(1),
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

# DAG Definition
dag = DAG(
    'data_science_etl_pipeline',
    default_args=default_args,
    description='ETL Pipeline для Data Science Analytics',
    schedule_interval='@daily',
    catchup=False,
    tags=['data-science', 'etl', 'analytics'],
)

# Logging konfigurieren
logger = logging.getLogger(__name__)

def extract_customer_data(**context):
    """Kundendaten aus verschiedenen Quellen extrahieren"""
    logger.info("Extrahiere Kundendaten...")
    
    # PostgreSQL Hook
    postgres_hook = PostgresHook(postgres_conn_id='postgres_default')
    
    # Kundendaten aus PostgreSQL extrahieren
    customer_query = """
        SELECT 
            customer_id,
            first_name,
            last_name,
            email,
            phone,
            registration_date,
            last_login_date,
            total_orders,
            total_spent,
            customer_segment
        FROM customers 
        WHERE last_login_date >= DATE_SUB(CURRENT_DATE, INTERVAL 30 DAY)
    """
    
    customers_df = postgres_hook.get_pandas_df(customer_query)
    logger.info(f"{len(customers_df)} Kundendatensätze extrahiert")
    
    # S3 Hook für zusätzliche Daten
    s3_hook = S3Hook(aws_conn_id='aws_default')
    
    # Produkt-Interaktionsdaten aus S3
    bucket_name = Variable.get('data_bucket_name', default_var='analytics-data')
    
    try:
        # CSV-Datei aus S3 laden
        file_content = s3_hook.read_key(key='customer_interactions.csv', bucket_name=bucket_name)
        interactions_df = pd.read_csv(pd.StringIO(file_content))
        logger.info(f"{len(interactions_df)} Interaktionsdatensätze aus S3 geladen")
    except Exception as e:
        logger.warning(f"Keine Interaktionsdaten in S3 gefunden: {e}")
        interactions_df = pd.DataFrame()
    
    # Daten kombinieren
    if not interactions_df.empty:
        combined_df = customers_df.merge(
            interactions_df, 
            on='customer_id', 
            how='left'
        )
    else:
        combined_df = customers_df
    
    # Daten als JSON speichern für nächste Task
    combined_json = combined_df.to_json(date_format='iso')
    
    # In XCom speichern
    task_instance = context['task_instance']
    task_instance.xcom_push(key='customer_data', value=combined_json)
    
    logger.info("Kundendaten-Extraktion abgeschlossen")
    return combined_json

def transform_customer_data(**context):
    """Kundendaten transformieren und bereinigen"""
    logger.info("Transformiere Kundendaten...")
    
    # Daten aus vorheriger Task holen
    task_instance = context['task_instance']
    customer_json = task_instance.xcom_pull(task_ids='extract_customer_data', key='customer_data')
    
    # JSON zu DataFrame
    customers_df = pd.read_json(customer_json)
    
    # Data Cleaning
    logger.info("Führe Data Cleaning durch...")
    
    # 1. Duplikate entfernen
    original_count = len(customers_df)
    customers_df = customers_df.drop_duplicates(subset=['customer_id'])
    duplicates_removed = original_count - len(customers_df)
    logger.info(f"{duplicates_removed} Duplikate entfernt")
    
    # 2. Fehlende Werte behandeln
    # Numerische Spalten: Median imputieren
    numeric_columns = ['total_orders', 'total_spent']
    for col in numeric_columns:
        if col in customers_df.columns:
            median_value = customers_df[col].median()
            customers_df[col].fillna(median_value, inplace=True)
    
    # Kategoriale Spalten: Mode imputieren
    categorical_columns = ['customer_segment']
    for col in categorical_columns:
        if col in customers_df.columns:
            mode_value = customers_df[col].mode()[0] if not customers_df[col].mode().empty else 'Unknown'
            customers_df[col].fillna(mode_value, inplace=True)
    
    # 3. Datums-Spalten normalisieren
    date_columns = ['registration_date', 'last_login_date']
    for col in date_columns:
        if col in customers_df.columns:
            customers_df[col] = pd.to_datetime(customers_df[col], errors='coerce')
    
    # 4. Feature Engineering
    logger.info("Erstelle neue Features...")
    
    # Customer Lifetime Value (CLV) berechnen
    if 'total_spent' in customers_df.columns and 'total_orders' in customers_df.columns:
        customers_df['avg_order_value'] = customers_df['total_spent'] / customers_df['total_orders']
        customers_df['clv_score'] = customers_df['total_spent'] * np.log1p(customers_df['total_orders'])
    
    # Recency Score (Tage seit letztem Login)
    if 'last_login_date' in customers_df.columns:
        current_date = datetime.now()
        customers_df['days_since_last_login'] = (current_date - customers_df['last_login_date']).dt.days
        customers_df['recency_score'] = pd.cut(customers_df['days_since_last_login'], 
                                              bins=[0, 7, 30, 90, float('inf')], 
                                              labels=[4, 3, 2, 1])
    
    # Engagement Score basierend auf verschiedenen Faktoren
    engagement_features = []
    if 'total_orders' in customers_df.columns:
        engagement_features.append('total_orders')
    if 'days_since_last_login' in customers_df.columns:
        engagement_features.append('days_since_last_login')
    
    if engagement_features:
        # Normalisierte Features für Engagement Score
        for feature in engagement_features:
            if feature in customers_df.columns:
                min_val = customers_df[feature].min()
                max_val = customers_df[feature].max()
                if max_val != min_val:
                    customers_df[f'{feature}_normalized'] = (customers_df[feature] - min_val) / (max_val - min_val)
                else:
                    customers_df[f'{feature}_normalized'] = 0
        
        # Engagement Score berechnen
        normalized_features = [f'{f}_normalized' for f in engagement_features]
        customers_df['engagement_score'] = customers_df[normalized_features].mean(axis=1)
    
    # 5. Datenvalidierung
    logger.info("Führe Datenvalidierung durch...")
    
    validation_errors = []
    
    # Business Rules prüfen
    if 'total_orders' in customers_df.columns:
        if (customers_df['total_orders'] < 0).any():
            validation_errors.append("Negative Bestellanzahlen gefunden")
    
    if 'total_spent' in customers_df.columns:
        if (customers_df['total_spent'] < 0).any():
            validation_errors.append("Negative Umsätze gefunden")
    
    if 'email' in customers_df.columns:
        invalid_emails = customers_df[~customers_df['email'].str.contains('@', na=False)]
        if not invalid_emails.empty:
            validation_errors.append(f"{len(invalid_emails)} ungültige E-Mail-Adressen gefunden")
    
    if validation_errors:
        logger.warning(f"Validierungsfehler: {validation_errors}")
    
    # 6. Daten aggregieren für Analytics
    logger.info("Aggregiere Daten für Analytics...")
    
    # Kunden-Segment Statistiken
    if 'customer_segment' in customers_df.columns:
        segment_stats = customers_df.groupby('customer_segment').agg({
            'customer_id': 'count',
            'total_spent': ['sum', 'mean'],
            'total_orders': ['sum', 'mean'],
            'engagement_score': 'mean'
        }).round(2)
        
        logger.info("Kunden-Segment Statistiken erstellt")
    
    # Transformierte Daten speichern
    transformed_json = customers_df.to_json(date_format='iso')
    task_instance.xcom_push(key='transformed_data', value=transformed_json)
    
    logger.info(f"Datentransformation abgeschlossen: {len(customers_df)} Datensätze")
    return transformed_json

def load_data_to_warehouse(**context):
    """Transformierte Daten in Data Warehouse laden"""
    logger.info("Lade Daten in Data Warehouse...")
    
    # Transformierte Daten holen
    task_instance = context['task_instance']
    transformed_json = task_instance.xcom_pull(task_ids='transform_customer_data', key='transformed_data')
    
    # JSON zu DataFrame
    customers_df = pd.read_json(transformed_json)
    
    # PostgreSQL Hook
    postgres_hook = PostgresHook(postgres_conn_id='postgres_default')
    
    # Tabelle erstellen falls nicht existent
    create_table_sql = """
    CREATE TABLE IF NOT EXISTS analytics_customer_metrics (
        customer_id VARCHAR(50) PRIMARY KEY,
        first_name VARCHAR(100),
        last_name VARCHAR(100),
        email VARCHAR(255),
        phone VARCHAR(50),
        registration_date TIMESTAMP,
        last_login_date TIMESTAMP,
        total_orders INTEGER,
        total_spent DECIMAL(10,2),
        customer_segment VARCHAR(50),
        avg_order_value DECIMAL(10,2),
        clv_score DECIMAL(10,2),
        days_since_last_login INTEGER,
        recency_score INTEGER,
        engagement_score DECIMAL(5,2),
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
        updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    );
    """
    
    postgres_hook.run(create_table_sql)
    
    # Upsert Operation (Insert oder Update)
    upsert_sql = """
    INSERT INTO analytics_customer_metrics (
        customer_id, first_name, last_name, email, phone, 
        registration_date, last_login_date, total_orders, total_spent,
        customer_segment, avg_order_value, clv_score, days_since_last_login,
        recency_score, engagement_score
    ) VALUES (
        %(customer_id)s, %(first_name)s, %(last_name)s, %(email)s, %(phone)s,
        %(registration_date)s, %(last_login_date)s, %(total_orders)s, %(total_spent)s,
        %(customer_segment)s, %(avg_order_value)s, %(clv_score)s, %(days_since_last_login)s,
        %(recency_score)s, %(engagement_score)s
    )
    ON CONFLICT (customer_id) 
    DO UPDATE SET
        first_name = EXCLUDED.first_name,
        last_name = EXCLUDED.last_name,
        email = EXCLUDED.email,
        phone = EXCLUDED.phone,
        registration_date = EXCLUDED.registration_date,
        last_login_date = EXCLUDED.last_login_date,
        total_orders = EXCLUDED.total_orders,
        total_spent = EXCLUDED.total_spent,
        customer_segment = EXCLUDED.customer_segment,
        avg_order_value = EXCLUDED.avg_order_value,
        clv_score = EXCLUDED.clv_score,
        days_since_last_login = EXCLUDED.days_since_last_login,
        recency_score = EXCLUDED.recency_score,
        engagement_score = EXCLUDED.engagement_score,
        updated_at = CURRENT_TIMESTAMP;
    """
    
    # Daten in Batches laden (Performance-Optimierung)
    batch_size = 1000
    total_rows = len(customers_df)
    
    for i in range(0, total_rows, batch_size):
        batch_df = customers_df.iloc[i:i+batch_size]
        
        # Daten für PostgreSQL vorbereiten
        batch_records = batch_df.to_dict('records')
        
        # NaN-Werte behandeln
        for record in batch_records:
            for key, value in record.items():
                if pd.isna(value):
                    record[key] = None
                elif isinstance(value, pd.Timestamp):
                    record[key] = value.isoformat()
        
        # Batch ausführen
        postgres_hook.run(upsert_sql, parameters=batch_records)
        
        logger.info(f"Batch {i//batch_size + 1}/{(total_rows-1)//batch_size + 1} geladen: {len(batch_df)} Zeilen")
    
    # BigQuery auch laden für Analytics
    try:
        bigquery_hook = BigQueryHook(gcp_conn_id='google_cloud_default', use_legacy_sql=False)
        
        # Dataset und Table konfigurieren
        project_id = Variable.get('gcp_project_id', default_var='my-project')
        dataset_id = 'analytics'
        table_id = 'customer_metrics'
        
        # DataFrame zu BigQuery laden
        customers_df.to_gbq(
            destination_table=f'{dataset_id}.{table_id}',
            project_id=project_id,
            if_exists='replace',
            progress_bar=False
        )
        
        logger.info("Daten erfolgreich nach BigQuery geladen")
        
    except Exception as e:
        logger.warning(f"BigQuery Load fehlgeschlagen: {e}")
    
    logger.info("Daten-Ladevorgang abgeschlossen")

def generate_analytics_report(**context):
    """Analytics Report generieren"""
    logger.info("Generiere Analytics Report...")
    
    # PostgreSQL Hook für finale Daten
    postgres_hook = PostgresHook(postgres_conn_id='postgres_default')
    
    # Analytics Queries ausführen
    analytics_queries = {
        'customer_segments': """
            SELECT 
                customer_segment,
                COUNT(*) as customer_count,
                AVG(total_spent) as avg_total_spent,
                AVG(total_orders) as avg_total_orders,
                AVG(clv_score) as avg_clv_score,
                AVG(engagement_score) as avg_engagement_score
            FROM analytics_customer_metrics
            GROUP BY customer_segment
            ORDER BY avg_total_spent DESC
        """,
        
        'top_customers': """
            SELECT 
                customer_id,
                first_name,
                last_name,
                total_spent,
                total_orders,
                clv_score,
                engagement_score
            FROM analytics_customer_metrics
            ORDER BY clv_score DESC
            LIMIT 10
        """,
        
        'engagement_distribution': """
            SELECT 
                CASE 
                    WHEN engagement_score >= 0.8 THEN 'High'
                    WHEN engagement_score >= 0.6 THEN 'Medium'
                    WHEN engagement_score >= 0.4 THEN 'Low'
                    ELSE 'Very Low'
                END as engagement_level,
                COUNT(*) as customer_count,
                AVG(total_spent) as avg_total_spent
            FROM analytics_customer_metrics
            GROUP BY engagement_level
            ORDER BY avg_total_spent DESC
        """,
        
        'daily_trends': """
            SELECT 
                DATE(last_login_date) as login_date,
                COUNT(*) as active_customers,
                AVG(total_spent) as avg_spend_per_customer
            FROM analytics_customer_metrics
            WHERE last_login_date >= CURRENT_DATE - INTERVAL 30 DAY
            GROUP BY DATE(last_login_date)
            ORDER BY login_date DESC
        """
    }
    
    # Ergebnisse sammeln
    analytics_results = {}
    
    for query_name, query in analytics_queries.items():
        try:
            result_df = postgres_hook.get_pandas_df(query)
            analytics_results[query_name] = result_df
            logger.info(f"Analytics Query '{query_name}' ausgeführt: {len(result_df)} Zeilen")
        except Exception as e:
            logger.error(f"Fehler bei Query '{query_name}': {e}")
            analytics_results[query_name] = pd.DataFrame()
    
    # Report erstellen
    report_data = {
        'generated_at': datetime.now().isoformat(),
        'total_customers': len(analytics_results.get('customer_segments', pd.DataFrame())),
        'analytics': {}
    }
    
    for query_name, result_df in analytics_results.items():
        if not result_df.empty:
            report_data['analytics'][query_name] = result_df.to_dict('records')
    
    # Report als JSON speichern
    report_json = json.dumps(report_data, indent=2, default=str)
    
    # Report in S3 speichern
    try:
        s3_hook = S3Hook(aws_conn_id='aws_default')
        bucket_name = Variable.get('reports_bucket_name', default_var='analytics-reports')
        
        report_filename = f"customer_analytics_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
        s3_hook.load_string(
            report_json,
            key=report_filename,
            bucket_name=bucket_name,
            replace=True
        )
        
        logger.info(f"Analytics Report gespeichert: s3://{bucket_name}/{report_filename}")
        
    except Exception as e:
        logger.error(f"Fehler beim Speichern des Reports: {e}")
    
    # Report-Metriken loggen
    if 'customer_segments' in analytics_results:
        segments_df = analytics_results['customer_segments']
        if not segments_df.empty:
            logger.info("Kunden-Segment Verteilung:")
            for _, row in segments_df.iterrows():
                logger.info(f"  {row['customer_segment']}: {row['customer_count']} Kunden, "
                           f"Ø Umsatz: €{row['avg_total_spent']:.2f}")
    
    logger.info("Analytics Report generierung abgeschlossen")

# Tasks definieren
extract_task = PythonOperator(
    task_id='extract_customer_data',
    python_callable=extract_customer_data,
    dag=dag,
)

transform_task = PythonOperator(
    task_id='transform_customer_data',
    python_callable=transform_customer_data,
    dag=dag,
)

load_task = PythonOperator(
    task_id='load_data_to_warehouse',
    python_callable=load_data_to_warehouse,
    dag=dag,
)

report_task = PythonOperator(
    task_id='generate_analytics_report',
    python_callable=generate_analytics_report,
    dag=dag,
)

# Task Dependencies
extract_task >> transform_task >> load_task >> report_task

3. Визуализация данных с Matplotlib, Seaborn и Plotly

import matplotlib.pyplot as plt
import seaborn as sns
import plotly.express as px
import plotly.graph_objects as go
from plotly.subplots import make_subplots
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import warnings
warnings.filterwarnings('ignore')

# Data Visualization Demo
class DataVisualizationDemo:
    
    def __init__(self):
        """Установить стиль и конфигурацию"""
        # Конфигурация Matplotlib
        plt.style.use('seaborn-v0_8')
        plt.rcParams['figure.figsize'] = (12, 8)
        plt.rcParams['font.size'] = 10
        plt.rcParams['axes.titlesize'] = 14
        plt.rcParams['axes.labelsize'] = 12
        plt.rcParams['xtick.labelsize'] = 10
        plt.rcParams['ytick.labelsize'] = 10
        plt.rcParams['legend.fontsize'] = 10
        plt.rcParams['figure.titlesize'] = 16
        
        # Конфигурация Seaborn
        sns.set_palette("husl")
        
        # Определить палитру цветов
        self.colors = {
            'primary': '#2E86AB',
            'secondary': '#A23B72',
            'accent': '#F18F01',
            'success': '#C73E1D',
            'warning': '#F4A261',
            'info': '#264653',
            'light': '#E9C46A',
            'dark': '#2A9D8F'
        }
        
        print("Data Visualization Demo инициализирована")
    
    def create_sample_data(self):
        """Создать примеры данных для визуализации"""
        np.random.seed(42)
        
        # Временные ряды (развитие доходов)
        dates = pd.date_range(start='2023-01-01', end='2023-12-31', freq='D')
        base_revenue = 10000
        trend = np.linspace(0, 5000, len(dates))
        seasonal = 2000 * np.sin(2 * np.pi * np.arange(len(dates)) / 365.25)
        noise = np.random.normal(0, 500, len(dates))
        
        revenue_data = pd.DataFrame({
            'date': dates,
            'revenue': base_revenue + trend + seasonal + noise,
            'orders': np.random.poisson(100, len(dates)) + 50,
            'customers': np.random.poisson(80, len(dates)) + 30
        })
        
        # Данные клиентов
        customer_data = pd.DataFrame({
            'customer_id': range(1000),
            'age': np.random.normal(35, 10, 1000),
            'income': np.random.lognormal(10.5, 0.5, 1000),
            'total_spent': np.random.gamma(2, 100, 1000),
            'orders': np.random.poisson(5, 1000) + 1,
            'segment': np.random.choice(['Bronze', 'Silver', 'Gold', 'Platinum'], 1000, 
                                      p=[0.4, 0.3, 0.2, 0.1]),
            'registration_date': pd.date_range('2020-01-01', '2023-12-31', periods=1000)[np.random.permutation(1000)],
            'last_purchase': pd.date_range('2023-01-01', '2023-12-31', periods=1000)[np.random.permutation(1000)]
        })
        
        # Данные продуктов
        products = ['Laptop', 'Smartphone', 'Tablet', 'Headphones', 'Mouse', 'Keyboard', 'Monitor', 'Webcam']
        product_data = pd.DataFrame({
            'product': np.random.choice(products, 5000),
            'category': np.random.choice(['Electronics', 'Accessories'], 5000),
            'price': np.random.uniform(10, 1000, 5000),
            'quantity_sold': np.random.poisson(10, 5000) + 1,
            'rating': np.random.uniform(3, 5, 5000),
            'month': np.random.choice(range(1, 13), 5000)
        })
        
        # Географические данные
        regions = ['North', 'South', 'East', 'West', 'Central']
        geo_data = pd.DataFrame({
            'region': np.random.choice(regions, 1000),
            'city': [f"{region} City {i}" for region, i in zip(
                np.random.choice(regions, 1000), 
                np.random.randint(1, 10, 1000)
            )],
            'population': np.random.lognormal(10, 1, 1000),
            'revenue': np.random.gamma(2, 50000, 1000),
            'stores': np.random.poisson(5, 1000) + 1
        })
        
        return revenue_data, customer_data, product_data, geo_data
    
    def create_time_series_visualizations(self, revenue_data):
        """Создать визуализации временных рядов"""
        print("Создание визуализаций временных рядов...")
        
        fig, axes = plt.subplots(2, 2, figsize=(16, 12))
        fig.suptitle('Анализ доходов во времени', fontsize=16, fontweight='bold')
        
        # 1. Развитие доходов
        axes[0, 0].plot(revenue_data['date'], revenue_data['revenue'], 
                       color=self.colors['primary'], linewidth=2)
        axes[0, 0].set_title('Ежедневное развитие доходов')
        axes[0, 0].set_xlabel('Дата')
        axes[0, 0].set_ylabel('Доход (€)')
        axes[0, 0].grid(True, alpha=0.3)
        
        # Добавить скользящие средние
        revenue_data['ma_7'] = revenue_data['revenue'].rolling(window=7).mean()
        revenue_data['ma_30'] = revenue_data['revenue'].rolling(window=30).mean()
        
        axes[0, 0].plot(revenue_data['date'], revenue_data['ma_7'], 
                       color=self.colors['accent'], linewidth=1, alpha=0.7, label='7-день MA')
        axes[0, 0].plot(revenue_data['date'], revenue_data['ma_30'], 
                       color=self.colors['secondary'], linewidth=1, alpha=0.7, label='30-день MA')
        axes[0, 0].legend()
        
        # 2. Месячный доход
        monthly_revenue = revenue_data.set_index('date').resample('M')['revenue'].sum()
        axes[0, 1].bar(monthly_revenue.index, monthly_revenue.values, 
                      color=self.colors['info'], alpha=0.7)
        axes[0, 1].set_title('Общий месячный доход')
        axes[0, 1].set_xlabel('Месяц')
        axes[0, 1].set_ylabel('Доход (€)')
        axes[0, 1].tick_params(axis='x', rotation=45)
        
        # 3. Доход против заказов, диаграмма рассеяния
        axes[1, 0].scatter(revenue_data['orders'], revenue_data['revenue'], 
                          alpha=0.6, color=self.colors['primary'])
        axes[1, 0].set_title('Доход против заказов')
        axes[1, 0].set_xlabel('Количество заказов')
        axes[1, 0].set_ylabel('Доход (€)')
        
        # Добавить линию тренда
        z = np.polyfit(revenue_data['orders'], revenue_data['revenue'], 1)
        p = np.poly1d(z)
        axes[1, 0].plot(revenue_data['orders'], p(revenue_data['orders']), 
                       color=self.colors['accent'], linewidth=2)
        
        # 4. Распределение ежедневных доходов
        axes[1, 1].hist(revenue_data['revenue'], bins=30, color=self.colors['secondary'], 
                       alpha=0.7, edgecolor='black')
        axes[1, 1].set_title('Распределение ежедневных доходов')
        axes[1, 1].set_xlabel('Доход (€)')
        axes[1, 1].set_ylabel('Частота')
        axes[1, 1].axvline(revenue_data['revenue'].mean(), color=self.colors['accent'], 
                           linestyle='--', linewidth=2, label=f'Среднее: €{revenue_data["revenue"].mean():.0f}')
        axes[1, 1].legend()
        
        plt.tight_layout()
        plt.savefig('time_series_analysis.png', dpi=300, bbox_inches='tight')
        plt.show()
        
        return fig
    
    def create_customer_analytics_visualizations(self, customer_data):
        """Создать визуализации аналитики клиентов"""
        print("Создание визуализаций аналитики клиентов...")
        
        fig, axes = plt.subplots(2, 3, figsize=(18, 12))
        fig.suptitle('Панель управления аналитикой клиентов', fontsize=16, fontweight='bold')
        
        # 1. Распределение по возрасту
        axes[0, 0].hist(customer_data['age'], bins=20, color=self.colors['primary'], 
                       alpha=0.7, edgecolor='black')
        axes[0, 0].set_title('Распределение по возрасту')
        axes[0, 0].set_xlabel('Возраст')
        axes[0, 0].set_ylabel('Количество')
        
        # 2. Распределение доходов
        axes[0, 1].hist(customer_data['income'], bins=30, color=self.colors['secondary'], 
                       alpha=0.7, edgecolor='black')
        axes[0, 1].set_title('Распределение доходов')
        axes[0, 1].set_xlabel('Доход (€)')
        axes[0, 1].set_ylabel('Количество')
        
        # 3. Распределение по сегментам клиентов
        segment_counts = customer_data['segment'].value_counts()
        axes[0, 2].pie(segment_counts.values, labels=segment_counts.index, 
                      autopct='%1.1f%%', colors=[self.colors['primary'], self.colors['secondary'], 
                                                self.colors['accent'], self.colors['success']])
        axes[0, 2].set_title('Распределение по сегментам')
        
        # 4. Доход по сегментам
        segment_revenue = customer_data.groupby('segment')['total_spent'].mean()
        bars = axes[1, 0].bar(segment_revenue.index, segment_revenue.values, 
                              color=[self.colors['primary'], self.colors['secondary'], 
                                     self.colors['accent'], self.colors['success']])
        axes[1, 0].set_title('Средний доход по сегментам')
        axes[1, 0].set_xlabel('Сегмент')
        axes[1, 0].set_ylabel('Ø Доход (€)')
        
        # Показать значения на столбцах
        for bar, value in zip(bars, segment_revenue.values):
            axes[1, 0].text(bar.get_x() + bar.get_width()/2, bar.get_height() + 100,
                           f'€{value:.0f}', ha='center', va='bottom')
        
        # 5. Возраст против общих расходов
        scatter = axes[1, 1].scatter(customer_data['age'], customer_data['total_spent'], 
                                   alpha=0.6, c=customer_data['orders'], cmap='viridis')
        axes[1, 1].set_title('Возраст против расходов')
        axes[1, 1].set_xlabel('Возраст')
        axes[1, 1].set_ylabel('Общие расходы (€)')
        plt.colorbar(scatter, ax=axes[1, 1], label='Количество заказов')
        
        # 6. График размаха для доходов по сегментам
        segments = customer_data['segment'].unique()
        data_for_boxplot = [customer_data[customer_data['segment'] == seg]['total_spent'] 
                           for seg in segments]
        
        box_plot = axes[1, 2].boxplot(data_for_boxplot, labels=segments, patch_artist=True)
        colors = [self.colors['primary'], self.colors['secondary'], 
                 self.colors['accent'], self.colors['success']]
        
        for patch, color in zip(box_plot['boxes'], colors):
            patch.set_facecolor(color)
            patch.set_alpha(0.7)
        
        axes[1, 2].set_title('Распределение расходов по сегментам')
        axes[1, 2].set_xlabel('Сегмент')
        axes[1, 2].set_ylabel('Общие расходы (€)')
        
        plt.tight_layout()
        plt.savefig('customer_analytics.png', dpi=300, bbox_inches='tight')
        plt.show()
        
        return fig
    
    def create_product_analytics_visualizations(self, product_data):
        """Создать визуализации аналитики продуктов"""
        print("Создание визуализаций аналитики продуктов...")
        
        fig, axes = plt.subplots(2, 2, figsize=(16, 12))
        fig.suptitle('Панель управления аналитикой продуктов', fontsize=16, fontweight='bold')
        
        # 1. Лучшие продукты по доходам
        product_revenue = product_data.groupby('product').apply(
            lambda x: (x['price'] * x['quantity_sold']).sum()
        ).sort_values(ascending=False)
        
        bars = axes[0, 0].barh(product_revenue.index, product_revenue.values, 
                              color=self.colors['primary'])
        axes[0, 0].set_title('Лучшие продукты по доходу')
        axes[0, 0].set_xlabel('Общий доход (€)')
        
        # Показать значения на столбцах
        for i, (bar, value) in enumerate(zip(bars, product_revenue.values)):
            axes[0, 0].text(value + max(product_revenue.values) * 0.01, i,
                           f'€{value:,.0f}', va='center')
        
        # 2. Распределение цен по категориям
        categories = product_data['category'].unique()
        for i, category in enumerate(categories):
            category_data = product_data[product_data['category'] == category]['price']
            axes[0, 1].hist(category_data, bins=20, alpha=0.7, label=category,
                           color=self.colors['primary'] if i == 0 else self.colors['secondary'])
        
        axes[0, 1].set_title('Распределение цен по категориям')
        axes[0, 1].set_xlabel('Цена (€)')
        axes[0, 1].set_ylabel('Частота')
        axes[0, 1].legend()
        
        # 3. Оценка против цены
        axes[1, 0].scatter(product_data['price'], product_data['rating'], 
                          alpha=0.6, color=self.colors['accent'])
        axes[1, 0].set_title('Оценка против цены')
        axes[1, 0].set_xlabel('Цена (€)')
        axes[1, 0].set_ylabel('Оценка')
        
        # Коэффициент корреляции
        correlation = np.corrcoef(product_data['price'], product_data['rating'])[0, 1]
        axes[1, 0].text(0.05, 0.95, f'Корреляция: {correlation:.3f}', 
                       transform=axes[1, 0].transAxes, va='top',
                       bbox=dict(boxstyle='round', facecolor='white', alpha=0.8))
        
        # 4. Ежемесячные продажи
        monthly_sales = product_data.groupby('month')['quantity_sold'].sum()
        month_names = ['Jan', 'Feb', 'Mar', 'Apr', 'May', 'Jun', 
                      'Jul', 'Aug', 'Sep', 'Oct', 'Nov', 'Dec']
        
        axes[1, 1].bar(monthly_sales.index, monthly_sales.values, 
                      color=self.colors['info'], alpha=0.7)
        axes[1, 1].set_title('Ежемесячные объемы продаж')
        axes[1, 1].set_xlabel('Месяц')
        axes[1, 1].set_ylabel('Объем продаж')
        axes[1, 1].set_xticks(range(1, 13))
        axes[1, 1].set_xticklabels(month_names, rotation=45)
        
        plt.tight_layout()
        plt.savefig('product_analytics.png', dpi=300, bbox_inches='tight')
        plt.show()
        
        return fig
    
    def create_geographic_visualizations(self, geo_data):
        """Создать географические визуализации"""
        print("Создание географических визуализаций...")
        
        fig, axes = plt.subplots(2, 2, figsize=(16, 12))
        fig.suptitle('Панель географической аналитики', fontsize=16, fontweight='bold')
        
        # 1. Доход по регионам
        region_revenue = geo_data.groupby('region')['revenue'].sum().sort_values(ascending=False)
        bars = axes[0, 0].bar(region_revenue.index, region_revenue.values, 
                              color=[self.colors['primary'], self.colors['secondary'], 
                                     self.colors['accent'], self.colors['success'], self.colors['info']])
        axes[0, 0].set_title('Общий доход по регионам')
        axes[0, 0].set_ylabel('Доход (€)')
        
        # Показать значения на столбцах
        for bar, value in zip(bars, region_revenue.values):
            axes[0, 0].text(bar.get_x() + bar.get_width()/2, bar.get_height() + max(region_revenue.values) * 0.01,
                           f'€{value:,.0f}', ha='center', va='bottom')
        
        # 2. Население против доходов
        axes[0, 1].scatter(geo_data['population'], geo_data['revenue'], 
                          alpha=0.6, s=geo_data['stores']*10, 
                          c=self.colors['primary'])
        axes[0, 1].set_title('Население против доходов')
        axes[0, 1].set_xlabel('Население')
        axes[0, 1].set_ylabel('Доход (€)')
        
        # 3. Плотность магазинов по регионам
        region_stores = geo_data.groupby('region')['stores'].sum()
        axes[1, 0].pie(region_stores.values, labels=region_stores.index, 
                      autopct='%1.1f%%', colors=[self.colors['primary'], self.colors['secondary'], 
                                                self.colors['accent'], self.colors['success'], self.colors['info']])
        axes[1, 0].set_title('Распределение магазинов по регионам')
        
        # 4. Доход на один магазин
        region_avg_revenue = geo_data.groupby('region').apply(
            lambda x: x['revenue'].sum() / x['stores'].sum()
        ).sort_values(ascending=False)
        
        bars = axes[1, 1].bar(region_avg_revenue.index, region_avg_revenue.values, 
                              color=self.colors['secondary'])
        axes[1, 1].set_title('Средний доход на магазин')
        axes[1, 1].set_ylabel('Ø Доход на магазин (€)')
        
        # Показать значения на столбцах
        for bar, value in zip(bars, region_avg_revenue.values):
            axes[1, 1].text(bar.get_x() + bar.get_width()/2, bar.get_height() + max(region_avg_revenue.values) * 0.01,
                           f'€{value:,.0f}', ha='center', va='bottom')
        
        plt.tight_layout()
        plt.savefig('geographic_analytics.png', dpi=300, bbox_inches='tight')
        plt.show()
        
        return fig
    
    def create_interactive_visualizations(self, revenue_data, customer_data, product_data):
        """Создать интерактивные визуализации с Plotly"""
        print("Создание интерактивных визуализаций...")
        
        # 1. Интерактивный временной ряд
        fig1 = make_subplots(
            rows=2, cols=1,
            subplot_titles=('Ежедневный доход', 'Ежемесячный доход'),
            vertical_spacing=0.1
        )
        
        # Ежедневный доход с скользящими средними
        fig1.add_trace(
            go.Scatter(
                x=revenue_data['date'],
                y=revenue_data['revenue'],
                mode='lines',
                name='Ежедневный доход',
                line=dict(color='blue', width=1)
            ),
            row=1, col=1
        )
        
        fig1.add_trace(
            go.Scatter(
                x=revenue_data['date'],
                y=revenue_data['ma_7'],
                mode='lines',
                name='7-день MA',
                line=dict(color='orange', width=2)
            ),
            row=1, col=1
        )
        
        # Ежемесячный доход
        monthly_revenue = revenue_data.set_index('date').resample('M')['revenue'].sum()
        fig1.add_trace(
            go.Bar(
                x=monthly_revenue.index,
                y=monthly_revenue.values,
                name='Ежемесячный доход',
                marker_color='lightblue'
            ),
            row=2, col=1
        )
        
        fig1.update_layout(
            title='Интерактивный анализ доходов',
            height=600,
            showlegend=True
        )
        
        fig1.show()
        
        # 2. Интерактивная аналитика клиентов
        fig2 = make_subplots(
            rows=2, cols=2,
            subplot_titles=('Распределение по возрасту', 'Распределение доходов', 
                          'Доход по сегментам', 'Возраст против доходов'),
            specs=[[{"type": "histogram"}, {"type": "histogram"}],
                   [{"type": "bar"}, {"type": "scatter"}]]
        )
        
        # Распределение по возрасту
        fig2.add_trace(
            go.Histogram(
                x=customer_data['age'],
                nbinsx=20,
                name='Возраст',
                marker_color='lightgreen'
            ),
            row=1, col=1
        )
        
        # Распределение доходов
        fig2.add_trace(
            go.Histogram(
                x=customer_data['income'],
                nbinsx=30,
                name='Доход',
                marker_color='lightcoral'
            ),
            row=1, col=2
        )
        
        # Доход по сегментам
        segment_revenue = customer_data.groupby('segment')['total_spent'].mean()
        fig2.add_trace(
            go.Bar(
                x=segment_revenue.index,
                y=segment_revenue.values,
                name='Ø Доход',
                marker_color='lightblue'
            ),
            row=2, col=1
        )
        
        # Возраст против доходов
        fig2.add_trace(
            go.Scatter(
                x=customer_data['age'],
                y=customer_data['total_spent'],
                mode='markers',
                name='Клиенты',
                marker=dict(
                    size=8,
                    color=customer_data['orders'],
                    colorscale='Viridis',
                    showscale=True,
                    colorbar=dict(title="Заказы")
                )
            ),
            row=2, col=2
        )
        
        fig2.update_layout(
            title='Интерактивная аналитика клиентов',
            height=600,
            showlegend=False
        )
        
        fig2.show()
        
        # 3. Интерактивная 3D визуализация
        fig3 = go.Figure(data=[go.Scatter3d(
            x=customer_data['age'],
            y=customer_data['income'],
            z=customer_data['total_spent'],
            mode='markers',
            marker=dict(
                size=5,
                color=customer_data['orders'],
                colorscale='Viridis',
                showscale=True,
                colorbar=dict(title="Заказы")
            ),
            text=[f"Клиент {i}<br>Возраст: {age}<br>Доход: {inc:,.0f}€<br>Доход: {spent:,.0f}€" 
                   for i, (age, inc, spent) in enumerate(zip(customer_data['age'], 
                                                          customer_data['income'], 
                                                          customer_data['total_spent']))],
            hovertemplate='%{text}<extra></extra>'
        )])
        
        fig3.update_layout(
            title='3D аналитика клиентов (Возраст, Доход, Расходы)',
            scene=dict(
                xaxis_title='Возраст',
                yaxis_title='Доход (€)',
                zaxis_title='Доход (€)'
            ),
            height=600
        )
        
        fig3.show()
        
        return fig1, fig2, fig3
    
    def create_dashboard_summary(self, revenue_data, customer_data, product_data, geo_data):
        """Создать сводную панель управления"""
        print("Создание сводной панели управления...")
        
        # Рассчитать ключевые показатели
        total_revenue = revenue_data['revenue'].sum()
        avg_daily_revenue = revenue_data['revenue'].mean()
        total_customers = len(customer_data)
        avg_customer_value = customer_data['total_spent'].mean()
        total_orders = customer_data['orders'].sum()
        top_product = product_data.groupby('product').apply(
            lambda x: (x['price'] * x['quantity_sold']).sum()
        ).idxmax()
        
        # Создать панель управления
        fig = plt.figure(figsize=(20, 16))
        
        # Заголовок
        fig.suptitle('Data Science Analytics Dashboard', fontsize=20, fontweight='bold')
        
        # Ключевые показатели
        metrics_text = f"""
        KEY METRICS
        ─────────────────────
        Общий доход: €{total_revenue:,.0f}
        Ø Ежедневный доход: €{avg_daily_revenue:,.0f}
        Всего клиентов: {total_customers:,}
        Ø Стоимость клиента: €{avg_customer_value:,.0f}
        Всего заказов: {total_orders:,}
        Лучший продукт: {top_product}
        """
        
        plt.figtext(0.02, 0.95, metrics_text, fontsize=12, fontfamily='monospace',
                   verticalalignment='top', bbox=dict(boxstyle='round', facecolor='lightgray', alpha=0.8))
        
        # Субграфики для разных анализов
        gs = fig.add_gridspec(3, 3, hspace=0.3, wspace=0.3, 
                           left=0.15, right=0.95, top=0.85, bottom=0.05)
        
        # 1. Тренд доходов (вверху слева)
        ax1 = fig.add_subplot(gs[0, 0])
        monthly_revenue = revenue_data.set_index('date').resample('M')['revenue'].sum()
        ax1.plot(monthly_revenue.index, monthly_revenue.values, 
                color=self.colors['primary'], linewidth=2)
        ax1.set_title('Тренд месячных доходов')
        ax1.tick_params(axis='x', rotation=45)
        
        # 2. Сегменты клиентов (вверху посередине)
        ax2 = fig.add_subplot(gs[0, 1])
        segment_counts = customer_data['segment'].value_counts()
        ax2.pie(segment_counts.values, labels=segment_counts.index, autopct='%1.1f%%',
               colors=[self.colors['primary'], self.colors['secondary'], 
                      self.colors['accent'], self.colors['success']])
        ax2.set_title('Сегменты клиентов')
        
        # 3. Лучшие продукты (вверху справа)
        ax3
## Рабочий процесс в Data Science

### Типичный процесс Data Science
```mermaid
graph TD
    A[Problem Definition] --> B[Data Collection]
    B --> C[Data Cleaning]
    C --> D[Exploratory Analysis]
    D --> E[Feature Engineering]
    E --> F[Model Building]
    F --> G[Model Evaluation]
    G --> H[Deployment]
    H --> I[Monitoring]
    I --> A

Методология CRISP-DM

1. Business Understanding: изучение бизнес-целей
2. Data Understanding: исследование данных
3. Data Preparation: очистка и подготовка данных
4. Modeling: разработка моделей
5. Evaluation: оценка моделей
6. Deployment: развёртывание моделей
7. Monitoring: мониторинг производительности

Технологии Big Data

Экосистема Hadoop

Компонент	Функция	Альтернативы
HDFS	Распределённое хранилище	Amazon S3, Google Cloud Storage
MapReduce	Пакетная обработка	Apache Spark, Apache Flink
Hive	SQL на Hadoop	Presto, Apache Impala
HBase	NoSQL база данных	Cassandra, MongoDB
Zookeeper	Координация	Consul, etcd

Spark vs. MapReduce

Функция	Spark	MapReduce
Скорость	In-Memory	Disk-based
Удобство	High APIs	Low-level
Streaming	Native	Batch only
Поддержка ML	MLlib	External
Использование ресурсов	Высокое	Низкое

Компоненты Machine Learning Pipeline

Этапы Data Pipeline

# ML Pipeline Beispiel mit Scikit-learn
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.compose import ColumnTransformer
from sklearn.ensemble import RandomForestRegressor
from sklearn.model_selection import train_test_split

# Preprocessing
numeric_features = ['age', 'income', 'orders']
categorical_features = ['segment', 'region']

numeric_transformer = Pipeline(steps=[
    ('scaler', StandardScaler())
])

categorical_transformer = Pipeline(steps=[
    ('onehot', OneHotEncoder(handle_unknown='ignore'))
])

preprocessor = ColumnTransformer(
    transformers=[
        ('num', numeric_transformer, numeric_features),
        ('cat', categorical_transformer, categorical_features)
    ])

# Complete Pipeline
ml_pipeline = Pipeline(steps=[
    ('preprocessor', preprocessor),
    ('regressor', RandomForestRegressor(n_estimators=100))
])

Практики MLOps

• Version Control: Git для кода, DVC для данных
• Experiment Tracking: MLflow, Weights & Biases
• Model Registry: MLflow Model Registry
• CI/CD: GitHub Actions, Jenkins
• Monitoring: Prometheus, Grafana
• Drift Detection: Evidently AI, NannyML

Принципы визуализации данных

Руководство по выбору графиков

Тип данных	Визуализация	Назначение
Временной ряд	Линейный график	Тренды во времени
Категориальные	Столбчатая диаграмма	Сравнения
Распределение	Гистограмма	Частотность
Взаимосвязь	Диаграмма рассеяния	Корреляции
Композиция	Круговая диаграмма	Доли
Географические	Карта	Локации

Теория цвета в визуализации данных

• Основные цвета: синий, красный, жёлтый
• Вторичные цвета: зелёный, оранжевый, фиолетовый
• Цветовая психология: синий = доверие, красный = опасность, зелёный = успех
• Контрастность: светлое и тёмное для читаемости
• Цветовая слепота: 8-10% населения

Best Practices в ETL

Проверки качества данных

def validate_data_quality(df):
    """Datenqualität validieren"""
    quality_report = {
        'total_rows': len(df),
        'missing_values': df.isnull().sum().to_dict(),
        'duplicate_rows': df.duplicated().sum(),
        'data_types': df.dtypes.to_dict(),
        'outliers': detect_outliers(df)
    }
    return quality_report

def detect_outliers(df, column):
    """Outliers mit IQR-Methode erkennen"""
    Q1 = df[column].quantile(0.25)
    Q3 = df[column].quantile(0.75)
    IQR = Q3 - Q1
    lower_bound = Q1 - 1.5 * IQR
    upper_bound = Q3 + 1.5 * IQR
    outliers = df[(df[column] < lower_bound) | (df[column] > upper_bound)]
    return len(outliers)

Оптимизация производительности

• Параллельная обработка: Multiprocessing, Dask
• Управление памятью: Chunking, Generators
• Индексирование: Database Indexes
• Кэширование: Redis, Memcached
• Пакетная обработка: Bulk Operations

Техники аналитики

Статистические методы

• Описательная статистика: Mean, Median, Mode, Std Dev
• Инференциальная статистика: Hypothesis Testing, Confidence Intervals
• Корреляционный анализ: Pearson, Spearman, Kendall
• Регрессионный анализ: Linear, Logistic, Polynomial
• Временные ряды: ARIMA, Prophet, LSTM

Продвинутая аналитика

• Кластеризация: K-Means, Hierarchical, DBSCAN
• Классификация: Decision Trees, Random Forest, SVM
• Выявление аномалий: Isolation Forest, One-Class SVM
• Снижение размерности: PCA, t-SNE, UMAP
• Правила ассоциации: Apriori, FP-Growth

Преимущества и недостатки

Преимущества Data Science

• Data-Driven Decisions: лучшие деловые решения
• Pattern Recognition: выявление скрытых закономерностей
• Predictive Analytics: прогнозирование будущих тенденций
• Process Optimization: повышение эффективности
• Competitive Advantage: конкурентное преимущество

Недостатки

• Data Quality: зависимость от качества данных
• Complexity: сложность алгоритмов и инструментов
• Privacy Concerns: вопросы конфиденциальности и этики
• Resource Intensive: требовательность к вычислительным ресурсам и памяти
• Interpretability: интерпретируемость черноящиковых моделей

Часто встречаемые вопросы на экзаменах

1. В чём разница между Big Data и традиционными данными? Big Data характеризуется тремя V (Volume, Velocity, Variety) и требует специализированных технологий для обработки.

2. Объясните процесс ETL! ETL расшифровывается как Extract (извлечение данных), Transform (очистка и преобразование) и Load (загрузка в целевую систему).

3. Когда какой тип графика использовать? Временные ряды для трендов, столбчатые диаграммы для сравнений, гистограммы для распределений, диаграммы рассеяния для взаимосвязей.

4. В чём назначение Machine Learning Pipeline? ML-пайплайны автоматизируют и стандартизируют весь процесс от подготовки данных до развёртывания модели.

Главные источники

1. https://spark.apache.org/
2. https://airflow.apache.org/
3. https://matplotlib.org/
4. https://plotly.com/python/