Специалист по анализу больших данных (Профессиональный уровень)

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("CountryStats") \
    .getOrCreate()

# Чтение JSON-файла
df = spark.read.json("users.json")

# Получение уникальных стран
unique_countries = df.select("country").distinct().orderBy("country")

# Подсчет количества стран
count = unique_countries.count()

print(f"Количество уникальных стран: {count}")
unique_countries.show(truncate=False)

Экзаменационный билет №4

Теоретическая часть:

1. Что такое schema evolution и как он поддерживается в Big Data системах?
2. Какие типы NoSQL баз данных существуют и когда их использовать?

Ответы на теоретическую часть:

1. Schema evolution — это изменение структуры данных без потери совместимости. Поддерживается форматами Avro, Parquet, ORC, которые позволяют добавлять, удалять поля и менять типы. Это критично при работе с историческими данными и streaming pipeline'ами.

2. Типы NoSQL:

   • Документоориентированные (MongoDB) — для гибкой структуры, например, профили пользователей.
   • Ключ-значение (Redis) — для кэширования и быстрого доступа.
   • Столбцовые (Cassandra) — для high-write нагрузок и аналитики.
   • Графовые (Neo4j) — для анализа связей и отношений. Выбор зависит от масштабируемости, задержки, сложности данных и требований к согласованности.

Практическая часть:

3. Напишите Spark-приложение на Scala, которое читает файл logs.txt, считает частоту встречаемости слов и выводит топ-10 самых популярных слов.

import org.apache.spark.sql.SparkSession

object WordCount {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder
      .appName("Word Count")
      .getOrCreate()

    val textFile = spark.sparkContext.textFile("logs.txt")
    
    val counts = textFile.flatMap(line => line.split(" "))
                         .map(word => (word.toLowerCase, 1))
                         .reduceByKey(_ + _)
                         .map(item => item.swap)
                         .sortByKey(false)

    counts.take(10).foreach(println)
  }
}

Экзаменационный билет №5

Теоретическая часть:

1. Какие существуют методы очистки и подготовки данных перед анализом в Big Data?
2. Как реализуется dynamic resource allocation в Spark и как её настроить?

Ответы на теоретическую часть:

1. Методы очистки и подготовки:

   • Удаление дубликатов
   • Заполнение пропусков (imputation)
   • Нормализация и стандартизация
   • Кодирование категориальных переменных (one-hot, label encoding)
   • Детектирование и обработка выбросов
   • Стандартизация форматов Инструменты: Pandas, PySpark, OpenRefine

2. Dynamic resource allocation позволяет Spark запрашивать и освобождать executors в зависимости от нагрузки. Настройка:

   spark.dynamicAllocation.enabled=true
   spark.dynamicAllocation.initialExecutors=2
   spark.dynamicAllocation.maxExecutors=20
   spark.dynamicAllocation.minExecutors=1
   spark.dynamicAllocation.localityWait=3s

   Это повышает использование ресурсов и снижает затраты при переменной нагрузке.

Практическая часть:

3. Напишите Spark SQL-запрос, который объединяет две таблицы: orders (order_id, user_id, amount) и users (user_id, name), выбирает имя пользователя и сумму заказа, и фильтрует заказы с суммой больше 500.

SELECT u.name, o.amount
FROM orders o
JOIN users u ON o.user_id = u.user_id
WHERE o.amount > 500;

Экзаменационный билет №6

Теоретическая часть:

1. Что такое ACID и какие Big Data системы его поддерживают?
2. Какие существуют стратегии масштабирования Big Data систем и в чём их отличие?

Ответы на теоретическую часть:

1. ACID — это свойства транзакций, обеспечивающие Atomicity (атомарность), Consistency (согласованность), Isolation (изолированность) и Durability (долговечность). В Big Data эти свойства поддерживаются такими системами, как Delta Lake, Apache Hudi, Apache Iceberg, HBase, Cassandra. Они критически важны для обеспечения точности и надежности обработки данных.

2. Стратегии масштабирования:

   • Горизонтальное масштабирование (scale-out) — добавление новых узлов в кластер. Это гибкий и отказоустойчивый подход, часто используемый в распределённых системах.
   • Вертикальное масштабирование (scale-up) — увеличение мощности существующих узлов (например, памяти или CPU). Ограничен физическими возможностями оборудования. Горизонтальное масштабирование предпочтительнее в Big Data из-за своей эластичности и отказоустойчивости.

Практическая часть:

3. Напишите PySpark-скрипт, который считывает CSV-файл transactions.csv, группирует данные по полю category, рассчитывает сумму amount и максимальное значение timestamp. Результат должен быть отсортирован по сумме в порядке убывания и сохранен в формате Parquet.

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("TransactionAggregation") \
    .getOrCreate()

# Чтение CSV
df = spark.read.csv("transactions.csv", header=True, inferSchema=True)

# Группировка и агрегация
result_df = df.groupBy("category") \
              .agg({"amount": "sum", "timestamp": "max"}) \
              .withColumnRenamed("sum(amount)", "total_amount") \
              .withColumnRenamed("max(timestamp)", "latest_timestamp")

# Сортировка и запись в Parquet
result_df.orderBy("total_amount", ascending=False) \
         .write.parquet("output/transaction_summary")

Экзаменационный билет №7

Теоретическая часть:

1. Что такое data governance и зачем она нужна в Big Data?
2. Какие методы используются для снижения латентности в real-time pipeline’ах?

Ответы на теоретическую часть:

1. Data governance — это набор политик, стандартов и процедур для обеспечения целостности, безопасности и соответствия использования данных. Она необходима для соблюдения регуляторных требований (например, GDPR), повышения доверия к данным и улучшения взаимодействия между командами.

2. Методы снижения латентности:

   • Micro-batch processing (в Spark Structured Streaming)
   • In-memory computing
   • Caching промежуточных результатов
   • Оптимизация state store
   • Настройка backpressure
   • Аппаратное ускорение (GPU/FPGA) Эти подходы позволяют сократить время ответа и повысить эффективность потоковой обработки.

Практическая часть:

3. Напишите Spark SQL запрос, который выбирает пользователей из таблицы users, у которых есть хотя бы один заказ из таблицы orders, и выводит имя пользователя, количество заказов и среднюю сумму заказа.

SELECT u.name, COUNT(o.order_id) AS order_count, AVG(o.amount) AS avg_amount
FROM users u
JOIN orders o ON u.user_id = o.user_id
GROUP BY u.name;

Экзаменационный билет №8

Теоретическая часть:

1. Что такое shuffle в Spark и почему он важен?
2. Как реализуется predicate pushdown и как он ускоряет обработку данных?

Ответы на теоретическую часть:

1. Shuffle — это процесс перераспределения данных между executors при операциях типа groupByKey или reduceByKey. Он требует значительных I/O и сетевых ресурсов, поэтому минимизация shuffle критична для производительности Spark-приложений.

2. Predicate pushdown — это перенос фильтров на уровень источника данных, чтобы уменьшить объём передаваемых данных. Реализован в Spark, Hive, Presto и других системах. Позволяет снизить I/O и ускорить выполнение запросов.

Практическая часть:

3. Напишите Scala-скрипт для Spark, который читает файл logs.txt, разделяет каждую строку на слова, подсчитывает частоту каждого слова и сохраняет результат в файл word_counts.csv.

import org.apache.spark.SparkContext
import org.apache.spark.SparkConf

object WordCount {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("Word Count")
    val sc = new SparkContext(conf)

    val textFile = sc.textFile("logs.txt")
    val counts = textFile.flatMap(line => line.split(" "))
                         .map(word => word.toLowerCase())
                         .map(word => (word, 1))
                         .reduceByKey(_ + _)

    counts.map(item => s"${item._1},${item._2}")
          .saveAsTextFile("word_counts.csv")
  }
}

Экзаменационный билет №9

Теоретическая часть:

1. Как работают bloom filters и где они применяются в Big Data?
2. Какие особенности работы с semi-structured и unstructured данными?

Ответы на теоретическую часть:

1. Bloom filter — это вероятностная структура данных, которая проверяет принадлежность элемента множеству. Применяется в HBase, Cassandra, Parquet для ускорения поиска и фильтрации, особенно при чтении из больших наборов данных. Используется для минимизации дискового I/O.

2. Полуструктурированные данные (JSON, XML) имеют внутреннюю иерархию, но не фиксированную схему. Неструктурированные (тексты, изображения) требуют feature extraction, NLP, computer vision. Обработка осуществляется через Spark, Hive, HCatalog, JSON SerDe.

Практическая часть:

3. Напишите PySpark-скрипт, который загружает JSON-файл events.json, фильтрует события с типом click, группирует по user_id и считает общее количество кликов на пользователя.

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("ClickCounter") \
    .getOrCreate()

# Чтение JSON
df = spark.read.json("events.json")

# Фильтрация и группировка
clicks_per_user = df.filter(df.event_type == "click") \
                    .groupBy("user_id") \
                    .count() \
                    .withColumnRenamed("count", "click_count")

# Сохранение результата
clicks_per_user.write.csv("output/clicks_per_user", header=True)

Экзаменационный билет №10

Теоретическая часть:

1. Какие метрики важны при оценке эффективности обработки данных?
2. Как происходит интеграция Big Data решений с бизнес-приложениями?

Ответы на теоретическую часть:

1. Ключевые метрики:

   • Время выполнения задачи
   • Стоимость ресурсов
   • Качество выходных данных (полнота, точность)
   • Производительность чтения/записи
   • Уровень отказоустойчивости
   • Latency и throughput Также важна повторяемость и воспроизводимость результатов.

2. Интеграция Big Data решений с бизнес-приложениями осуществляется через API (REST/gRPC), ESB (Enterprise Service Bus), message brokers (Kafka, RabbitMQ), JDBC/ODBC-коннекторы. Также возможна интеграция с BI-системами, CRM, ERP и другими корпоративными решениями.

Практическая часть:

3. Напишите SQL-запрос на Spark SQL, который выбирает пользователей из таблицы users, у которых более 3 заказов в таблице orders, и выводит их имя и количество заказов.

SELECT u.name, COUNT(o.order_id) AS order_count
FROM users u
JOIN orders o ON u.user_id = o.user_id
GROUP BY u.name
HAVING COUNT(o.order_id) > 3;

Экзаменационный билет №11

Теоретическая часть:

1. Что такое data pipeline и как его спроектировать?
2. Какие существуют best practices для управления качеством данных?

Ответы на теоретическую часть:

1. Data pipeline — это последовательность этапов обработки данных от источника до потребителя. Проектирование включает:

   • Определение источников и стока
   • Выбор технологий (Spark, Kafka, Airflow)
   • Этапы преобразования данных
   • Обработка ошибок и логирование
   • Мониторинг и масштабируемость

2. Best practices по управлению качеством:

   • Валидация на этапе ingest
   • Мониторинг дрейфа данных
   • Автоматическое тестирование
   • Документирование схем
   • Профилирование данных
   • Установка SLA по качеству
   • Интеграция с системами data governance

Практическая часть:

3. Напишите PySpark-скрипт, который читает файл users.csv, фильтрует пользователей старше 30 лет, группирует их по колонке country и считает количество пользователей в каждой стране. Результат сохранить в формате JSON.

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("UserStats") \
    .getOrCreate()

# Чтение CSV
df = spark.read.csv("users.csv", header=True, inferSchema=True)

# Фильтрация и группировка
result_df = df.filter(df.age > 30) \
              .groupBy("country") \
              .count() \
              .withColumnRenamed("count", "user_count")

# Сохранение результата в JSON
result_df.write.json("output/user_stats_by_country")

Экзаменационный билет №12

Теоретическая часть:

1. Какие подходы используются для очистки и подготовки данных перед анализом?
2. Как реализуется shuffle partitioning и как он влияет на производительность Spark?

Ответы на теоретическую часть:

1. Подходы к очистке и подготовке:

   • Удаление дубликатов
   • Заполнение пропусков (imputation)
   • Нормализация и стандартизация
   • Кодирование категориальных переменных
   • Обнаружение и обработка выбросов
   • Приведение к единому формату Используются библиотеки: Pandas, PySpark, OpenRefine

2. Shuffle partitioning определяет, как данные распределяются между узлами при операциях groupBy, join и reduceByKey. Число партиций задаётся через параметр spark.sql.shuffle.partitions. Слишком малое число может вызвать дисбаланс нагрузки, слишком большое — увеличить overhead. Оптимальное значение зависит от объёма данных и ресурсов кластера.

Практическая часть:

3. Напишите SQL-запрос на Spark SQL, который выбирает из таблицы transactions только те записи, где сумма больше среднего значения по таблице, и выводит user_id, amount, timestamp.

SELECT user_id, amount, timestamp
FROM transactions
WHERE amount > (SELECT AVG(amount) FROM transactions);

Экзаменационный билет №13

Теоретическая часть:

1. Как организовать работу с графовыми алгоритмами в Spark GraphX?
2. Какие современные тенденции в области анализа больших данных?

Ответы на теоретическую часть:

1. GraphX предоставляет API для представления графов и реализации алгоритмов: PageRank, Connected Components, Shortest Paths, Triangle Counting. Работа включает создание RDD вершин и рёбер, применение Pregel API и оптимизацию партиционирования.

2. Современные тенденции:

   • Lakehouse architecture (Delta Lake, Iceberg)
   • MLOps и AIOps
   • Real-time analytics
   • Edge computing + streaming
   • Self-service BI
   • AI-driven ETL
   • Гибридные и многооблачные решения
   • Автоматизация feature engineering
   • Акцент на privacy и compliance

Практическая часть:

3. Напишите Scala-скрипт, который использует GraphX для создания простого графа из набора пар (srcId, dstId) и запускает алгоритм PageRank.

import org.apache.spark.graphx._
import org.apache.spark._

object GraphExample {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("GraphPageRank")
    val sc = new SparkContext(conf)

    // Создание RDD вершин (по умолчанию все имеют значение 1.0)
    val vertices = sc.parallelize(Seq((1L, 1.0), (2L, 1.0), (3L, 1.0)))

    // Создание ребер
    val edges = sc.parallelize(Seq(Edge(1L, 2L, 1.0), Edge(2L, 3L, 1.0), Edge(3L, 1L, 1.0)))

    // Создание графа
    val graph = Graph(vertices, edges)

    // Вычисление PageRank
    val ranks = graph.pageRank(0.001).vertices

    // Вывод результатов
    ranks.collect.foreach(println)
  }
}

Экзаменационный билет №14

Теоретическая часть:

1. Какие особенности работы с медиафайлами в Big Data системах?
2. Какие методы используются для снижения латентности в real-time pipeline’ах?

Ответы на теоретическую часть:

1. Медиаданные (видео, аудио, изображения) требуют специальной обработки: хранения в blob-хранилищах (S3, ADLS), извлечения метаданных, применения computer vision (OpenCV, TensorFlow), speech recognition (Whisper, Kaldi) и индексации через Elasticsearch или FAISS.

2. Методы снижения латентности:

   • Micro-batch processing (Spark Structured Streaming)
   • In-memory computing
   • Caching
   • Pre-aggregation
   • Backpressure tuning
   • State store optimization Эти подходы позволяют достичь low-latency обработки и быстрого ответа на события.

Практическая часть:

3. Напишите PySpark-скрипт, который читает поток событий из Kafka, фильтрует сообщения с ключом "error", и сохраняет их в Parquet-файлы.

from pyspark.sql import SparkSession
from pyspark.sql.functions import col

spark = SparkSession.builder \
    .appName("KafkaErrorLogger") \
    .getOrCreate()

# Чтение из Kafka
df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "host1:9092") \
    .option("subscribe", "logs") \
    .load()

# Преобразование и фильтрация
filtered_df = df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)") \
                .filter(col("key") == "error")

# Запись в Parquet
query = filtered_df.writeStream \
                   .outputMode("append") \
                   .format("parquet") \
                   .option("checkpointLocation", "checkpoint/error_log") \
                   .option("path", "output/errors") \
                   .start()

query.awaitTermination()

Экзаменационный билет №15

Теоретическая часть:

1. Как реализуется schema registry в streaming системах?
2. Как происходит работа с временным рядом в Big Data?

Ответы на теоретическую часть:

1. Schema Registry — централизованное хранилище схем данных. Используется в Kafka с Confluent Schema Registry для обеспечения совместимости сообщений, контроля версий и сериализации в Avro/Protobuf. Позволяет избежать несовместимых изменений и упрощает обработку потоковых данных.

2. Обработка временных рядов включает:

   • Windowed aggregations
   • Time-series databases (InfluxDB, OpenTSDB)
   • Прогнозирование (ARIMA, Prophet, LSTM)
   • Anomaly detection (Isolation Forest, Holt-Winters)
   • Использование Spark и Flink для распределённой обработки

Практическая часть:

3. Напишите Spark SQL запрос, который рассчитывает скользящее среднее за последние 7 дней по полю value из таблицы time_series_data.

SELECT ts, value,
       AVG(value) OVER (
           ORDER BY ts 
           ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
       ) AS moving_avg_7d
FROM time_series_data;