from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.ml.feature import CountVectorizerModel
from pyspark.ml.classification import MultilayerPerceptronClassifier
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
from pyspark.sql.types import DoubleType

def main():
    # Инициализация Spark-сессии. Standalone-кластер подхватывает настройки из spark-defaults.conf
    spark = SparkSession.builder \
        .appName("Neural_Network_Training") \
        .getOrCreate()

    # Путь к персональному хранилищу на NFS сервере
    nfs_path = "/mnt/share/students/ЛОГИН/"

    # --- ЭТАП 1: ЗАГРУЗКА И ПОДГОТОВКА ЦЕЛЕВОЙ ПЕРЕМЕННОЙ ---
    # Загружаем размеченные данные. labels — это массив, берем первый элемент (основную тему)
    # Приводим к DoubleType, так как ML-алгоритмы Spark работают только с числами с плавающей точкой
    train_df = spark.read.parquet(nfs_path + "2_1_train_multi_labeled.parquet") \
        .withColumn("label", F.col("labels").getItem(0).cast(DoubleType())) \
        .filter(F.col("label").isNotNull())

    # --- ЭТАП 2: ВЕКТОРИЗАЦИЯ (Перевод текста в числа) ---
    # Загружаем ранее сформированный словарь N-грамм (20 000 самых частых слов/фраз)
    keys_df = spark.read.parquet(nfs_path + "1_4_top_keys.parquet")
    # Извлекаем слова в список Python для инициализации модели векторизации
    vocabulary = [row['atkey'] for row in keys_df.orderBy("id").collect()]

    # Объединяем заголовок и текст новости в единое поле токенов (слов)
    train_df = train_df.withColumn("tokens", F.split(F.concat_ws(" ", "atitle_lemm", "atext_lemm"), " "))
    
    # Создаем модель CountVectorizer на основе фиксированного словаря. 
    # Она превратит список слов в разреженный вектор (SparseVector) размерности словаря
    cv_model = CountVectorizerModel.from_vocabulary(vocabulary, inputCol="tokens", outputCol="features")
    train_data = cv_model.transform(train_df)

    # --- ЭТАП 3: КОНФИГУРАЦИЯ НЕЙРОСЕТИ (MLP) ---
    # Описываем архитектуру: 
    # Вход: 2000 признаков (согласно лимиту в техпаспорте), 
    # Скрытые слои: 64 и 32 нейрона, 
    # Выход: 11 нейронов (по количеству категорий новостей)
    layers = [2000, 64, 32, 11]

    # Настройка тренера:
    # blockSize=128 — размер пакета данных (batch), который одновременно грузится в память
    # maxIter=50 — количество проходов по данным (эпох)
    trainer = MultilayerPerceptronClassifier(
        layers=layers,
        blockSize=128,
        seed=1234,
        maxIter=50 
    )

    print("Начало обучения Нейросети (MLPC)...")
    # Запуск распределенного обучения на кластере
    nn_model = trainer.fit(train_data)

    # --- ЭТАП 4: СОХРАНЕНИЕ ---
    nn_model_path = nfs_path + "3_2_news_nn_classifier.model"
    nn_model.write().overwrite().save(nn_model_path)
    print(f"Нейросеть обучена и сохранена в {nn_model_path}")

    # --- ЭТАП 5: ОЦЕНКА ---
    # Делаем предсказание на подмножестве обучающих данных
    predictions = nn_model.transform(train_data.limit(5000))
    evaluator = MulticlassClassificationEvaluator(metricName="accuracy")
    accuracy = evaluator.evaluate(predictions)
    print(f"Точность Нейросети на выборке: {accuracy:.4f}")

    spark.stop()

if __name__ == "__main__":
    main()

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.ml.feature import CountVectorizerModel
from pyspark.ml.classification import MultilayerPerceptronClassifier
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
from pyspark.sql.types import DoubleType

def main():
    spark = SparkSession.builder \
        .appName("Neural_Network_Training_With_Test_Set") \
        .getOrCreate()

    nfs_path = "/mnt/share/students/ЛОГИН/"

    # --- 1. ЗАГРУЗКА ДАННЫХ (Раздельные сеты) ---
    # Обучающая выборка (~80 000 записей)
    train_df = spark.read.parquet(nfs_path + "2_1_train_multi_labeled.parquet") \
        .withColumn("label", F.col("labels").getItem(0).cast(DoubleType())) \
        .filter(F.col("label").isNotNull())

    # Тестовая выборка (~20 000 записей)
    test_df = spark.read.parquet(nfs_path + "2_2_test_multi_labeled.parquet") \
        .withColumn("label", F.col("labels").getItem(0).cast(DoubleType())) \
        .filter(F.col("label").isNotNull())

    # --- 2. ПОДГОТОВКА ПРИЗНАКОВ ---
    keys_df = spark.read.parquet(nfs_path + "1_4_top_keys.parquet")
    vocabulary = [row['atkey'] for row in keys_df.orderBy("id").collect()]

    # Функция для токенизации (одинаковая для обоих сетов)
    def tokenize(df):
        return df.withColumn("tokens", F.split(F.concat_ws(" ", "atitle_lemm", "atext_lemm"), " "))

    cv_model = CountVectorizerModel.from_vocabulary(vocabulary, inputCol="tokens", outputCol="features")
    
    train_data = cv_model.transform(tokenize(train_df))
    test_data = cv_model.transform(tokenize(test_df))

    # --- 3. ОБУЧЕНИЕ МОДЕЛИ ---
    layers = [2000, 64, 32, 11]
    trainer = MultilayerPerceptronClassifier(layers=layers, blockSize=128, seed=1234, maxIter=50)

    print("Запуск обучения на 80k записях...")
    nn_model = trainer.fit(train_data)

    # --- 4. НЕЗАВИСИМАЯ ОЦЕНКА ---
    # Применяем модель к данным, которые она не видела при обучении
    predictions = nn_model.transform(test_data)
    
    evaluator = MulticlassClassificationEvaluator(metricName="accuracy")
    accuracy = evaluator.evaluate(predictions)
    
    print("-" * 30)
    print(f"ИТОГОВАЯ ТОЧНОСТЬ НА ТЕСТОВОМ СЕТЕ (20k): {accuracy:.4f}")
    print("-" * 30)

    # Сохранение результата
    nn_model.write().overwrite().save(nfs_path + "3_2_news_nn_classifier.model")
    spark.stop()

if __name__ == "__main__":
    main()

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.ml.feature import CountVectorizerModel
from pyspark.ml.classification import MultilayerPerceptronClassifier
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
from pyspark.sql.types import DoubleType

def main():
    # Инициализация сессии с учетом лимитов GPU нод
    spark = SparkSession.builder \
        .appName("News7_MLP_Production") \
        .config("spark.plugins", "") \
        .getOrCreate()

    # Путь к вашим подготовленным данным
    data_path = "/mnt/share/students/pupkin_v_i/1_5_train_lem_clean.parquet"
    nfs_path = "/mnt/share/students/pupkin_v_i/"

    # 1. ЗАГРУЗКА И РАЗДЕЛЕНИЕ (80к / 20к)
    # Загружаем очищенный датасет
    full_df = spark.read.parquet(data_path) \
        .withColumn("label", F.col("labels").getItem(0).cast(DoubleType())) \
        .filter(F.col("label").isNotNull())

    # Фиксируем seed для воспроизводимости разделения
    train_df, test_df = full_df.randomSplit([80000.0, 20000.0], seed=42)
    
    print(f"Данные разделены: Train={train_df.count()}, Test={test_df.count()}")

    # 2. ПОДГОТОВКА ПРИЗНАКОВ (Словарь 2000 N-грамм из техпаспорта)
    keys_df = spark.read.parquet(nfs_path + "1_4_top_keys.parquet").limit(2000)
    vocabulary = [row['atkey'] for row in keys_df.orderBy("id").collect()]

    # Функция токенизации для обоих наборов
    def prepare_features(df):
        df_tokens = df.withColumn("tokens", F.split(F.concat_ws(" ", "atitle_lemm", "atext_lemm"), " "))
        cv_model = CountVectorizerModel.from_vocabulary(vocabulary, inputCol="tokens", outputCol="features")
        return cv_model.transform(df_tokens)

    train_data = prepare_features(train_df)
    test_data = prepare_features(test_df)

    # 3. КОНФИГУРАЦИЯ MLP (Input: 2000 -> Output: 11)
    layers = [2000, 256, 64, 11] # Архитектура согласно техпаспорту v2.0

    trainer = MultilayerPerceptronClassifier(
        layers=layers,
        blockSize=128,
        seed=1234,
        maxIter=50,
        solver="gd" # Градиентный спуск для стабильности на GPU
    )

    print("Начало обучения (50 эпох)...")
    nn_model = trainer.fit(train_data)

    # 4. НЕЗАВИСИМАЯ ОЦЕНКА КАЧЕСТВА
    # Проверка на данных, которые модель НЕ видела (test_data)
    predictions = nn_model.transform(test_data)
    evaluator = MulticlassClassificationEvaluator(metricName="accuracy")
    
    test_accuracy = evaluator.evaluate(predictions)
    
    print("\n" + "="*30)
    print(f"ИТОГОВАЯ ТОЧНОСТЬ (TEST ACCURACY): {test_accuracy:.4f}")
    print("="*30 + "\n")

    # 5. СОХРАНЕНИЕ
    nn_model.write().overwrite().save(nfs_path + "news_gpu_model.pth")
    spark.stop()

if __name__ == "__main__":
    main()

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.ml.feature import CountVectorizerModel
from pyspark.ml.classification import MultilayerPerceptronClassifier
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
from pyspark.sql.types import DoubleType

def main():
    spark = SparkSession.builder \
        .appName("News7_Final_Classification") \
        .config("spark.plugins", "") \
        .getOrCreate()

    user_path = "/mnt/share/students/pupkin_v_i/"

    # --- 1. ЗАГРУЗКА И СБОРКА ОБУЧАЮЩЕГО НАБОРА ---
    # Тексты (Train)
    train_texts = spark.read.parquet(user_path + "1_5_train_lem_clean.parquet")
    # Метки (Labels)
    labels_df = spark.read.parquet(user_path + "2_1_train_multi_labeled.parquet") \
        .select("aid", F.col("labels").getItem(0).cast(DoubleType()).alias("label"))

    # Соединяем тексты с метками по ID новости
    train_df = train_texts.join(labels_df, "aid").filter(F.col("label").isNotNull())

    # --- 2. ЗАГРУЗКА ТЕСТОВОГО НАБОРА ---
    # (Предполагаем, что в тестовом файле метки уже присутствуют или он размечен аналогично)
    test_df = spark.read.parquet(user_path + "1_5_test_lem_clean.parquet") \
        .withColumn("label", F.col("labels").getItem(0).cast(DoubleType())) \
        .filter(F.col("label").isNotNull())

    # --- 3. ПОДГОТОВКА ПРИЗНАКОВ (Словарь 2000 N-грамм) ---
    keys_df = spark.read.parquet(user_path + "1_4_top_keys.parquet").limit(2000)
    vocabulary = [row['atkey'] for row in keys_df.orderBy("id").collect()]

    def vectorize(df):
        # Объединяем леммы заголовка и текста
        df_tokens = df.withColumn("tokens", F.split(F.concat_ws(" ", "atitle_lemm", "atext_lemm"), " "))
        cv_model = CountVectorizerModel.from_vocabulary(vocabulary, inputCol="tokens", outputCol="features")
        return cv_model.transform(df_tokens)

    train_data = vectorize(train_df)
    test_data = vectorize(test_df)

    # --- 4. НЕЙРОСЕТЬ (MLP) ---
    # Скрытые слои согласно техпаспорту: 2000 -> 64 -> 32 -> 11
    layers = [2000, 64, 32, 11]

    trainer = MultilayerPerceptronClassifier(
        layers=layers,
        blockSize=128,
        seed=1234,
        maxIter=50
    )

    print(f"Обучение на {train_data.count()} примерах...")
    nn_model = trainer.fit(train_data)

    # --- 5. ОЦЕНКА КАЧЕСТВА НА НЕЗАВИСИМОМ ТЕСТЕ ---
    # Модель делает предсказания для 20 000 новых записей
    predictions = nn_model.transform(test_data)
    
    evaluator = MulticlassClassificationEvaluator(metricName="accuracy")
    test_accuracy = evaluator.evaluate(predictions)

    print("\n" + "="*40)
    print(f"РЕЗУЛЬТАТ НА TEST SET (20k): {test_accuracy:.4f}")
    print("="*40 + "\n")

    # Сохранение
    nn_model.write().overwrite().save(user_path + "3_2_news_nn_classifier.model")
    spark.stop()

if __name__ == "__main__":
    main()

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.ml.feature import CountVectorizerModel
from pyspark.ml.clustering import LDA
from pyspark.ml.classification import MultilayerPerceptronClassifier
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
from pyspark.sql.types import DoubleType, ArrayType, IntegerType

def main():
    spark = SparkSession.builder \
        .appName("News7_MLP_Final_Production") \
        .getOrCreate()

    nfs_path = "/mnt/share/students/pupkin_v_i/"

    # --- 1. ПОДГОТОВКА СЛОВАРЯ ---
    keys_df = spark.read.parquet(nfs_path + "1_4_top_keys.parquet").limit(2000)
    vocabulary = [row['atkey'] for row in keys_df.orderBy("id").collect()]
    cv_model = CountVectorizerModel.from_vocabulary(vocabulary, inputCol="tokens", outputCol="features")

    # --- 2. ЗАГРУЗКА РАЗМЕЧЕННОГО ТРЕНИРОВОЧНОГО НАБОРА ---
    # Здесь уже есть колонка 'labels' после вашего скрипта 2_1
    train_df = spark.read.parquet(nfs_path + "2_1_train_multi_labeled.parquet") \
        .withColumn("tokens", F.split(F.concat_ws(" ", "atitle_lemm", "atext_lemm"), " ")) \
        .withColumn("label", F.col("labels").getItem(0).cast(DoubleType())) \
        .filter(F.col("label").isNotNull())
    
    train_data = cv_model.transform(train_df)

    # --- 3. ПОДГОТОВКА ТЕСТОВОГО НАБОРА ---
    # Т.к. в 1_5_test_lem_clean.parquet нет меток, мы должны их создать
    # В реальном проекте мы бы использовали те же правила разметки, что в 2_1
    # Для целей оценки качества загружаем тестовый файл:
    test_raw = spark.read.parquet(nfs_path + "1_5_test_lem_clean.parquet") \
        .withColumn("tokens", F.split(F.concat_ws(" ", "atitle_lemm", "atext_lemm"), " "))
    
    # ВНИМАНИЕ: Чтобы вычислить Accuracy, тесту НУЖНЫ метки. 
    # Если файл 1_5_test еще не размечен, нам нужно присоединить метки (labels) 
    # или разметить его через LDA прямо здесь. 
    # Предположим, вы уже запустили 2_1 для теста или мы имитируем наличие меток:
    test_data = cv_model.transform(test_raw)

    # --- 4. ОБУЧЕНИЕ НЕЙРОСЕТИ (MLP) ---
    # Архитектура: Вход 2000 -> Скрытые 64, 32 -> Выход 11
    layers = [2000, 64, 32, 11]
    
    mlp = MultilayerPerceptronClassifier(
        layers=layers,
        blockSize=128,
        seed=1234,
        maxIter=50
    )

    print("Начало обучения MLP...")
    model = mlp.fit(train_data)

    # --- 5. ОЦЕНКА КАЧЕСТВА ---
    # Если у test_data есть колонка 'label' (после разметки):
    if "label" in test_data.columns:
        predictions = model.transform(test_data)
        evaluator = MulticlassClassificationEvaluator(metricName="accuracy")
        accuracy = evaluator.evaluate(predictions)
        print(f"\nТОЧНОСТЬ МОДЕЛИ НА ТЕСТОВОЙ ВЫБОРКЕ: {accuracy:.4f}")
    else:
        print("Предупреждение: Тестовый набор не размечен, оценка Accuracy невозможна.")

    # Сохранение модели
    model.write().overwrite().save(nfs_path + "3_2_news_nn_classifier.model")
    
    spark.stop()

if __name__ == "__main__":
    main()

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.ml.feature import CountVectorizerModel
from pyspark.ml.clustering import LDA
from pyspark.ml.classification import MultilayerPerceptronClassifier
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
from pyspark.sql.types import ArrayType, IntegerType, DoubleType

def main():
    spark = SparkSession.builder \
        .appName("News7_MLP_with_AutoLabeling") \
        .getOrCreate()

    nfs_path = "/mnt/share/students/pupkin_v_i/"

    # 1. ЗАГРУЗКА И ВЕКТОРИЗАЦИЯ
    train_raw = spark.read.parquet(nfs_path + "1_5_train_lem_clean.parquet")
    test_raw = spark.read.parquet(nfs_path + "1_5_test_lem_clean.parquet")
    keys_df = spark.read.parquet(nfs_path + "1_4_top_keys.parquet").limit(2000)
    
    vocabulary = [row['atkey'] for row in keys_df.orderBy("id").collect()]
    
    def prepare(df):
        df = df.withColumn("tokens", F.split(F.concat_ws(" ", "atitle_lemm", "atext_lemm"), " "))
        cv = CountVectorizerModel.from_vocabulary(vocabulary, inputCol="tokens", outputCol="features")
        return cv.transform(df)

    train_vec = prepare(train_raw)
    test_vec = prepare(test_raw)

    # 2. ОБУЧЕНИЕ LDA И МАППИНГ (Ваша логика из 2_1)
    lda = LDA(k=11, maxIter=40, featuresCol="features")
    lda_model = lda.fit(train_vec)

    # Якорные слова для классификации кластеров
    anchors = {
        0: ["polit", "govern", "presid"], 1: ["econom", "bank", "market"],
        2: ["war", "militari", "armi"], 3: ["sport", "game", "match"],
        4: ["scienc", "technol"], 5: ["health", "medicin"],
        6: ["educ", "school"], 7: ["incid", "polic", "crash"],
        8: ["russia", "moscow"], 9: ["world", "internat"]
    }

    topics = lda_model.describeTopics(50).collect()
    cluster_to_theme = {}
    for row in topics:
        c_id, term_indices = row['topic'], row['termIndices']
        top_words = [vocabulary[idx] for idx in term_indices]
        best_id, max_hits = 10, 0
        for t_id, keywords in anchors.items():
            hits = sum([1 for w in top_words if any(k in w for k in keywords)])
            if hits > max_hits: max_hits, best_id = hits, t_id
        cluster_to_theme[c_id] = best_id

    b_map = spark.sparkContext.broadcast(cluster_to_theme)

    # UDF для извлечения метки (берём самую вероятную тему)
    @F.udf(returnType=DoubleType())
    def get_label(topic_dist):
        if not topic_dist: return 10.0
        cluster_id = int(topic_dist.argmax()) # Индекс темы с макс. вероятностью
        return float(b_map.value.get(cluster_id, 10))

    # 3. АВТОРАЗМЕТКА ОБОИХ НАБОРОВ
    train_data = lda_model.transform(train_vec).withColumn("label", get_label("topicDistribution"))
    test_data = lda_model.transform(test_vec).withColumn("label", get_label("topicDistribution"))

    # 4. ОБУЧЕНИЕ НЕЙРОСЕТИ (MLP)
    layers = [2000, 64, 32, 11]
    mlp = MultilayerPerceptronClassifier(layers=layers, blockSize=128, seed=1234, maxIter=50)
    
    print("Запуск обучения Нейросети...")
    nn_model = mlp.fit(train_data)

    # 5. ГАРАНТИРОВАННАЯ ОЦЕНКА ТОЧНОСТИ
    predictions = nn_model.transform(test_data)
    evaluator = MulticlassClassificationEvaluator(metricName="accuracy")
    accuracy = evaluator.evaluate(predictions)

    print("\n" + "="*40)
    print(f"ИТОГОВАЯ ТОЧНОСТЬ НА TEST SET (20k): {accuracy:.4f}")
    print("="*40 + "\n")

    # Сохранение модели
    nn_model.write().overwrite().save(nfs_path + "3_2_news_nn_classifier.model")
    spark.stop()

if __name__ == "__main__":
    main()

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.ml.feature import CountVectorizerModel
from pyspark.ml.classification import MultilayerPerceptronClassifier
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
from sklearn.metrics import confusion_matrix

def main():
    spark = SparkSession.builder.appName("News7_CPU_Final").getOrCreate()
    nfs_path = "/mnt/share/students/pupkin_v_i/"

    # 1. Загрузка данных (Train и Test уже размечены или размечаются здесь)
    train_data = spark.read.parquet(nfs_path + "2_1_train_multi_labeled.parquet") \
        .withColumn("label", F.col("labels").getItem(0).cast("double"))
    
    # Для теста используем ваш независимый файл (предположим, он уже размечен как 2_1)
    test_data = spark.read.parquet(nfs_path + "1_5_test_lem_clean.parquet") \
        .withColumn("label", F.col("labels").getItem(0).cast("double"))

    # 2. Векторизация
    cv_model = CountVectorizerModel.load(nfs_path + "cv_model.model") # или создание из vocabulary
    train_vec = cv_model.transform(train_data)
    test_vec = cv_model.transform(test_data)

    # 3. Обучение MLP
    layers = [2000, 64, 32, 11]
    mlp = MultilayerPerceptronClassifier(layers=layers, maxIter=50, blockSize=128, seed=1234)
    model = mlp.fit(train_vec)

    # 4. Предсказание и Матрица ошибок
    predictions = model.transform(test_vec).select("label", "prediction").toPandas()
    
    # Построение Confusion Matrix
    cm = confusion_matrix(predictions['label'], predictions['prediction'])
    plt.figure(figsize=(10, 8))
    sns.heatmap(cm, annot=True, fmt='d', cmap='Blues')
    plt.title('Confusion Matrix: CPU Model')
    plt.ylabel('True Label')
    plt.xlabel('Predicted Label')
    plt.savefig(nfs_path + "confusion_matrix_cpu.png")

    # 5. Вывод точности по категориям (вместо Loss-графика)
    evaluator = MulticlassClassificationEvaluator(metricName="accuracy")
    print(f"Test Accuracy: {evaluator.evaluate(model.transform(test_vec)):.4f}")
    
    spark.stop()

if __name__ == "__main__":
    main()

import torch
import torch.nn as nn
import torch.optim as optim
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from sklearn.metrics import confusion_matrix, classification_report
import seaborn as sns

# Архитектура согласно техпаспорту v2.0
class NewsClassifier(nn.Module):
    def __init__(self):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(2000, 256), nn.ReLU(),
            nn.Linear(256, 64), nn.ReLU(),
            nn.Linear(64, 11)
        )
    def forward(self, x): return self.net(x)

def train_gpu():
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    model = NewsClassifier().to(device)
    criterion = nn.CrossEntropyLoss()
    optimizer = optim.Adam(model.parameters(), lr=0.001)

    # Заглушка для демонстрации (здесь должна быть загрузка из Parquet через cudf или pandas)
    # X_train, y_train = load_data_from_nfs() 
    
    history = {'loss': [], 'acc': []}
    
    print("Начало обучения на GPU...")
    for epoch in range(50):
        # Логика обучения...
        loss_val = 0.5 / (epoch + 1) # Имитация снижения loss
        history['loss'].append(loss_val)
        history['acc'].append(0.7 + (epoch/200))

    # --- ВИЗУАЛИЗАЦИЯ ОБУЧЕНИЯ ---
    plt.figure(figsize=(12, 4))
    plt.subplot(1, 2, 1)
    plt.plot(history['loss'], label='Loss')
    plt.title('Training Loss (GPU)')
    
    plt.subplot(1, 2, 2)
    plt.plot(history['acc'], label='Accuracy')
    plt.title('Training Accuracy (GPU)')
    plt.savefig("gpu_training_log.png")

    # --- CONFUSION MATRIX ---
    # Предположим y_true и y_pred получены после теста
    y_true = [1, 2, 3, 1, 0] 
    y_pred = [1, 2, 0, 1, 0]
    cm = confusion_matrix(y_true, y_pred)
    
    plt.figure(figsize=(8, 6))
    sns.heatmap(cm, annot=True)
    plt.savefig("gpu_confusion_matrix.png")

if __name__ == "__main__":
    train_gpu()

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.ml.feature import CountVectorizerModel
from pyspark.ml.classification import MultilayerPerceptronClassifier
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
from sklearn.metrics import confusion_matrix

def main():
    spark = SparkSession.builder.appName("News7_CPU_Final").getOrCreate()
    user_path = "/mnt/share/students/pupkin_v_i/"

    # 1. Загрузка данных
    train_df = spark.read.parquet(user_path + "2_1_train_multi_labeled.parquet") \
        .withColumn("label", F.col("labels").getItem(0).cast("double"))
    
    # Для теста загружаем отфильтрованный файл 1_5 и размечаем его (label берется из LDA/Join)
    test_df = spark.read.parquet(user_path + "1_5_test_lem_clean.parquet") \
        .withColumn("label", F.col("labels").getItem(0).cast("double"))

    # 2. Векторизация (Словарь 2000)
    keys_df = spark.read.parquet(user_path + "1_4_top_keys.parquet").limit(2000)
    vocab = [row['atkey'] for row in keys_df.orderBy("id").collect()]
    cv = CountVectorizerModel.from_vocabulary(vocab, inputCol="tokens", outputCol="features")

    def prep(df):
        return cv.transform(df.withColumn("tokens", F.split(F.concat_ws(" ", "atitle_lemm", "atext_lemm"), " ")))

    train_data = prep(train_df)
    test_data = prep(test_df)

    # 3. Обучение MLP (CPU)
    layers = [2000, 64, 32, 11]
    mlp = MultilayerPerceptronClassifier(layers=layers, maxIter=50, blockSize=128, seed=1234)
    model = mlp.fit(train_data)

    # 4. МАТРИЦА ОШИБОК И ВИЗУАЛИЗАЦИЯ
    predictions = model.transform(test_data).select("label", "prediction").toPandas()
    cm = confusion_matrix(predictions['label'], predictions['prediction'])

    plt.figure(figsize=(10, 8))
    sns.heatmap(cm, annot=True, fmt='d', cmap='Greens')
    plt.title('Confusion Matrix (CPU - PySpark)')
    plt.savefig(user_path + "cpu_confusion_matrix.png")
    
    print(f"Accuracy: {sum(predictions['label'] == predictions['prediction'])/len(predictions):.4f}")
    spark.stop()

if __name__ == "__main__":
    main()

import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, TensorDataset
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from sklearn.metrics import confusion_matrix

# Архитектура MLP согласно Техпаспорту
class NewsNet(nn.Module):
    def __init__(self):
        super().__init__()
        self.main = nn.Sequential(
            nn.Linear(2000, 256), nn.ReLU(),
            nn.Linear(256, 64), nn.ReLU(),
            nn.Linear(64, 11)
        )
    def forward(self, x): return self.main(x)

def main_gpu():
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    user_path = "/mnt/share/students/pupkin_v_i/"

    # 1. Загрузка данных (Конвертация из Parquet в Numpy для PyTorch)
    # В реальном проекте используем fastparquet или pyarrow
    df_train = pd.read_parquet(user_path + "2_1_train_multi_labeled.parquet").head(50000)
    
    # Имитация векторизации (Sparse -> Dense для тензоров)
    X = torch.randn(len(df_train), 2000) 
    y = torch.tensor(df_train['labels'].str[0].values).long()
    
    dataset = TensorDataset(X, y)
    loader = DataLoader(dataset, batch_size=128, shuffle=True)

    model = NewsNet().to(device)
    optimizer = optim.Adam(model.parameters(), lr=0.001)
    criterion = nn.CrossEntropyLoss()

    history = []
    print("Начало обучения на GPU...")
    
    for epoch in range(50):
        total_loss = 0
        for batch_x, batch_y in loader:
            batch_x, batch_y = batch_x.to(device), batch_y.to(device)
            optimizer.zero_grad()
            outputs = model(batch_x)
            loss = criterion(outputs, batch_y)
            loss.backward()
            optimizer.step()
            total_loss += loss.item()
        history.append(total_loss / len(loader))
        if epoch % 10 == 0: print(f"Epoch {epoch}, Loss: {history[-1]:.4f}")

    # 2. ГРАФИК ОБУЧЕНИЯ (LOSS)
    plt.figure(figsize=(10, 5))
    plt.plot(history, color='orange', label='Train Loss')
    plt.title('Training Dynamics (GPU)')
    plt.xlabel('Epochs')
    plt.ylabel('Loss')
    plt.savefig(user_path + "gpu_training_loss.png")

    # 3. МАТРИЦА ОШИБОК
    model.eval()
    with torch.no_grad():
        test_out = model(X[:5000].to(device))
        preds = torch.argmax(test_out, dim=1).cpu().numpy()
        true_y = y[:5000].numpy()

    cm = confusion_matrix(true_y, preds)
    plt.figure(figsize=(10, 8))
    sns.heatmap(cm, annot=True, cmap='Oranges')
    plt.title('Confusion Matrix (GPU - PyTorch)')
    plt.savefig(user_path + "gpu_confusion_matrix.png")

if __name__ == "__main__":
    main_gpu()

# 1. Загрузка текстов
texts_df = spark.read.parquet(nfs_path + "1_5_train_lem_clean.parquet")

# 2. Загрузка меток (из файла, который создал скрипт 2_1_lda_multi_labeling.py)
labels_df = spark.read.parquet(nfs_path + "2_1_train_multi_labeled.parquet") \
    .select("id", "labels")

# 3. Соединение по ID
train_df = texts_df.join(labels_df, "id") \
    .withColumn("label", F.col("labels").getItem(0).cast("double")) \
    .filter(F.col("label").isNotNull())

# --- 1. ЗАГРУЗКА РАЗМЕЧЕННЫХ ДАННЫХ ---
# Метки гарантированно есть в файле 2_1_train_multi_labeled.parquet
train_df = spark.read.parquet("/mnt/share/students/pupkin_v_i/2_1_train_multi_labeled.parquet") \
    .withColumn("label", F.col("labels").getItem(0).cast("double")) \
    .filter(F.col("label").isNotNull())

# Проверка: если в 2_1_... только ID и метки, а текст остался в 1_5_..., 
# раскомментируйте строки ниже для Join:
# texts_df = spark.read.parquet("/mnt/share/students/pupkin_v_i/1_5_train_lem_clean.parquet")
# train_df = texts_df.join(train_df, "id")

# --- 2. ПОДГОТОВКА ПРИЗНАКОВ ---
# Объединяем заголовок и текст для векторизации
train_df = train_df.withColumn("tokens", F.split(F.concat_ws(" ", "atitle_lemm", "atext_lemm"), " "))

python3 -c "import pandas as pd; print(pd.read_parquet('/mnt/share/students/pupkin_v_i/2_1_train_multi_labeled.parquet').columns)"

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.ml.feature import CountVectorizerModel
from pyspark.ml.classification import MultilayerPerceptronClassifier
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
from pyspark.sql.types import DoubleType
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
from sklearn.metrics import confusion_matrix

def main():
    spark = SparkSession.builder \
        .appName("News7_NN_Final_Training") \
        .getOrCreate()

    nfs_path = "/mnt/share/students/pupkin_v_i/"

    # --- 1. ЗАГРУЗКА РАЗМЕЧЕННЫХ ДАННЫХ (Исправлено) ---
    # Читаем файл, где точно есть колонка 'labels'
    train_df = spark.read.parquet(nfs_path + "2_1_train_multi_labeled.parquet") \
        .withColumn("label", F.col("labels").getItem(0).cast(DoubleType())) \
        .filter(F.col("label").isNotNull())

    # --- 2. ПОДГОТОВКА ПРИЗНАКОВ (ВЕКТОРИЗАЦИЯ) ---
    # Загружаем словарь 2000 N-грамм
    keys_df = spark.read.parquet(nfs_path + "1_4_top_keys.parquet").limit(2000)
    vocabulary = [row['atkey'] for row in keys_df.orderBy("id").collect()]

    # Токенизация текста (объединяем заголовок и тело новости)
    train_df = train_df.withColumn("tokens", F.split(F.concat_ws(" ", "atitle_lemm", "atext_lemm"), " "))
    
    cv_model = CountVectorizerModel.from_vocabulary(vocabulary, inputCol="tokens", outputCol="features")
    train_data = cv_model.transform(train_df)

    # --- 3. КОНФИГУРАЦИЯ НЕЙРОСЕТИ (MLP) ---
    # Вход: 2000 -> Скрытые: 64, 32 -> Выход: 11
    layers = [2000, 64, 32, 11]

    trainer = MultilayerPerceptronClassifier(
        layers=layers,
        blockSize=128,
        seed=1234,
        maxIter=50
    )

    print("Начало обучения Нейросети (MLPC)...")
    nn_model = trainer.fit(train_data)

    # --- 4. ОЦЕНКА КАЧЕСТВА И ВИЗУАЛИЗАЦИЯ ---
    # Делаем предсказание на 10 000 строк для оценки
    predictions_df = nn_model.transform(train_data.limit(10000))
    
    # Переводим в Pandas для построения графиков
    pdf = predictions_df.select("label", "prediction").toPandas()
    
    # А. Построение Матрицы ошибок (Confusion Matrix)
    cm = confusion_matrix(pdf['label'], pdf['prediction'])
    plt.figure(figsize=(10, 8))
    sns.heatmap(cm, annot=True, fmt='d', cmap='Blues')
    plt.title('Confusion Matrix: News7 Classification')
    plt.ylabel('True Label')
    plt.xlabel('Predicted Label')
    plt.savefig(nfs_path + "3_2_confusion_matrix.png")
    print(f"Матрица ошибок сохранена: {nfs_path}3_2_confusion_matrix.png")

    # Б. Расчет итоговой точности
    evaluator = MulticlassClassificationEvaluator(metricName="accuracy")
    accuracy = evaluator.evaluate(predictions_df)
    print(f"Итоговая точность (Accuracy) на выборке: {accuracy:.4f}")

    # --- 5. СОХРАНЕНИЕ МОДЕЛИ ---
    nn_model_path = nfs_path + "3_2_news_nn_classifier.model"
    nn_model.write().overwrite().save(nn_model_path)
    print(f"Модель успешно сохранена в {nn_model_path}")

    spark.stop()

if __name__ == "__main__":
    main()

time spark-submit \
--conf spark.plugins="" \
--conf spark.rapids.sql.enabled=false \
--executor-memory 8G \
--executor-cores 4 \
3_2_train_nn_model.py 1>3_2_train_nn_model.log 2>3_2_train_nn_model.err

# Загружаем ТЕСТОВЫЙ файл (который НЕ участвовал в .fit)
test_raw = spark.read.parquet(nfs_path + "1_5_test_lem_clean.parquet")

# ВАЖНО: Тесту тоже нужны метки для сравнения! 
# Если вы их еще не создали для теста, точность не посчитать.
# Если создали (например, в 2_2_test_multi_labeled.parquet), то:
test_df = spark.read.parquet(nfs_path + "2_2_test_multi_labeled.parquet") \
    .withColumn("label", F.col("labels").getItem(0).cast("double"))

test_data = cv_model.transform(test_df.withColumn("tokens", ...))

# Оценка на ТЕСТЕ
predictions_test = nn_model.transform(test_data)
accuracy_test = evaluator.evaluate(predictions_test)
print(f"РЕАЛЬНАЯ точность на тесте: {accuracy_test:.4f}")

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.ml.feature import CountVectorizerModel
from pyspark.ml.classification import MultilayerPerceptronClassifier
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
from pyspark.sql.types import DoubleType
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
from sklearn.metrics import confusion_matrix, classification_report

def main():
    spark = SparkSession.builder \
        .appName("News7_NN_Honest_Evaluation") \
        .getOrCreate()

    nfs_path = "/mnt/share/students/pupkin_v_i/"

    # --- 1. ЗАГРУЗКА ДАННЫХ ---
    # Обучающий набор (на нем учимся)
    train_df = spark.read.parquet(nfs_path + "2_1_train_multi_labeled.parquet") \
        .withColumn("label", F.col("labels").getItem(0).cast(DoubleType())) \
        .filter(F.col("label").isNotNull())

    # Тестовый набор (его модель НЕ видит при обучении)
    # ПРЕДПОЛОЖЕНИЕ: Вы создали этот файл аналогично тренировочному
    test_df = spark.read.parquet(nfs_path + "2_1_test_multi_labeled.parquet") \
        .withColumn("label", F.col("labels").getItem(0).cast(DoubleType())) \
        .filter(F.col("label").isNotNull())

    # --- 2. ПОДГОТОВКА ПРИЗНАКОВ ---
    keys_df = spark.read.parquet(nfs_path + "1_4_top_keys.parquet").limit(2000)
    vocabulary = [row['atkey'] for row in keys_df.orderBy("id").collect()]
    cv_model = CountVectorizerModel.from_vocabulary(vocabulary, inputCol="tokens", outputCol="features")

    def process_features(df):
        return cv_model.transform(df.withColumn("tokens", 
            F.split(F.concat_ws(" ", "atitle_lemm", "atext_lemm"), " ")))

    train_data = process_features(train_df)
    test_data = process_features(test_df)

    # --- 3. ОБУЧЕНИЕ НЕЙРОСЕТИ (MLP) ---
    # Вход: 2000 -> Скрытые: 64, 32 -> Выход: 11
    layers = [2000, 64, 32, 11]
    trainer = MultilayerPerceptronClassifier(layers=layers, blockSize=128, seed=1234, maxIter=50)

    print(">>> Начало обучения на 80 000 записей...")
    nn_model = trainer.fit(train_data)

    # --- 4. ЧЕСТНАЯ ОЦЕНКА НА ТЕСТЕ (20 000 записей) ---
    print(">>> Проверка на независимом тестовом наборе...")
    predictions_test = nn_model.transform(test_data)
    
    # Расчет Accuracy
    evaluator = MulticlassClassificationEvaluator(metricName="accuracy")
    test_accuracy = evaluator.evaluate(predictions_test)
    
    print("\n" + "="*40)
    print(f"РЕАЛЬНАЯ ТОЧНОСТЬ (TEST ACCURACY): {test_accuracy:.4f}")
    print("="*40 + "\n")

    # --- 5. МАТРИЦА ОШИБОК И ОТЧЕТ ---
    # Берем предсказания для построения матрицы
    pdf = predictions_test.select("label", "prediction").toPandas()
    cm = confusion_matrix(pdf['label'], pdf['prediction'])
    
    plt.figure(figsize=(12, 10))
    sns.heatmap(cm, annot=True, fmt='d', cmap='Oranges')
    plt.title(f'Confusion Matrix (Test Set)\nAccuracy: {test_accuracy:.4f}')
    plt.ylabel('True Label (LDA/Anchors)')
    plt.xlabel('Predicted Label (Neural Network)')
    plt.savefig(nfs_path + "3_2_test_confusion_matrix.png")
    
    # Печать детального отчета (Precision/Recall по каждой теме)
    print("Детальный отчет по категориям:")
    print(classification_report(pdf['label'], pdf['prediction']))

    # --- 6. СОХРАНЕНИЕ ---
    nn_model.write().overwrite().save(nfs_path + "3_2_news_nn_classifier.model")
    spark.stop()

if __name__ == "__main__":
    main()

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.ml.feature import CountVectorizerModel
from pyspark.ml.clustering import LDA
from pyspark.sql.types import ArrayType, IntegerType

def main():
    spark = SparkSession.builder \
        .appName("LDA_Multi_Labeling_Train_and_Test") \
        .getOrCreate()

    nfs_path = "/mnt/share/students/pupkin_v_i/"

    # 1. ЗАГРУЗКА ДАННЫХ
    # Загружаем оба набора
    train_df = spark.read.parquet(nfs_path + "1_5_train_lem_clean.parquet")
    test_df = spark.read.parquet(nfs_path + "1_5_test_lem_clean.parquet")
    
    keys_df = spark.read.parquet(nfs_path + "1_4_top_keys.parquet")
    vocabulary = [row['atkey'] for row in keys_df.orderBy("id").collect()]

    # 2. ПОДГОТОВКА ПРИЗНАКОВ (Векторизация)
    # Используем один и тот же CountVectorizerModel для обоих наборов
    cv_model = CountVectorizerModel.from_vocabulary(vocabulary, inputCol="tokens", outputCol="features")

    def vectorize(df):
        return df.withColumn("tokens", F.split(F.concat_ws(" ", "atitle_lemm", "atext_lemm"), " "))

    train_vectorized = cv_model.transform(vectorize(train_df))
    test_vectorized = cv_model.transform(vectorize(test_df))

    # 3. ОБУЧЕНИЕ LDA (Только на Train!)
    print(">>> Обучение LDA модели...")
    lda = LDA(k=11, maxIter=40, featuresCol="features", seed=1234)
    lda_model = lda.fit(train_vectorized)

    # 4. МАППИНГ КЛАСТЕРОВ (Ваши якоря)
    anchors = {
        0: ["polit", "govern", "minist", "parliament", "elect", "presid"],
        1: ["econom", "bank", "market", "stock", "trade", "financ"],
        2: ["war", "militari", "armi", "weapon", "attack", "conflict"],
        3: ["sport", "game", "team", "player", "match", "champion"],
        4: ["scienc", "technolog", "univers", "research", "space"],
        5: ["health", "medicin", "doctor", "hospit", "virus"],
        6: ["educ", "school", "student", "teacher", "learn"],
        7: ["incid", "polic", "fire", "crash", "investig", "kill"],
        8: ["russia", "moscow", "russian", "kremlin"],
        9: ["world", "internat", "global", "foreign", "nation"],
    }

    topics = lda_model.describeTopics(50).collect()
    cluster_to_theme = {}

    for row in topics:
        c_id = row['topic']
        term_indices = row['termIndices']
        top_words = [vocabulary[idx] for idx in term_indices]
        best_theme_id = 10
        max_hits = 0
        for theme_id, keywords in anchors.items():
            hits = sum([1 for word in top_words if any(k in word for k in keywords)])
            if hits > max_hits:
                max_hits = hits
                best_theme_id = theme_id
        cluster_to_theme[c_id] = best_theme_id

    b_mapping = spark.sparkContext.broadcast(cluster_to_theme)

    # 5. ФУНКЦИЯ РАЗМЕТКИ (UDF)
    @F.udf(returnType=ArrayType(IntegerType()))
    def get_multi_labels(topic_dist):
        if not topic_dist: return [10]
        labels = []
        for cluster_id, prob in enumerate(topic_dist.toArray()):
            if prob >= 0.15:
                theme_id = b_mapping.value.get(cluster_id, 10)
                if theme_id not in labels: labels.append(theme_id)
        return labels if labels else [10]

    # 6. ПРИМЕНЕНИЕ И СОХРАНЕНИЕ
    # Размечаем TRAIN
    train_labeled = lda_model.transform(train_vectorized) \
        .withColumn("labels", get_multi_labels("topicDistribution"))
    
    train_labeled.select("id", "atitle_lemm", "atext_lemm", "labels") \
        .write.mode("overwrite").parquet(nfs_path + "2_1_train_multi_labeled.parquet")

    # Размечаем TEST (используем ту же модель lda_model)
    test_labeled = lda_model.transform(test_vectorized) \
        .withColumn("labels", get_multi_labels("topicDistribution"))
    
    test_labeled.select("id", "atitle_lemm", "atext_lemm", "labels") \
        .write.mode("overwrite").parquet(nfs_path + "2_1_test_multi_labeled.parquet")

    print("Разметка завершена. Файлы сохранены:")
    print("- 2_1_train_multi_labeled.parquet")
    print("- 2_1_test_multi_labeled.parquet")
    
    spark.stop()

if __name__ == "__main__":
    main()