Введение
Компания получает записи заказов и хочет превращать их в полезную информацию. Прежде чем рассчитывать отчёты продаж, нужен способ представить строки, описать столбцы и попросить инструмент обработки выполнить работу. Вы начнёте с трёх записей, которые легко проверить глазами.
Большие данные — нагрузки на данные, размер, скорость поступления или потребности обработки которых могут перерасти удобный подход на одной машине. Apache Spark — движок обработки данных, способный разделять работу между машинами кластера. PySpark — его интерфейс Python. DataFrame представляет строки с именованными типизированными столбцами и поддерживает операции, например фильтрацию и агрегацию. Spark обрабатывает данные; отдельный файл или система хранения обеспечивает постоянные входные и выходные данные. Курс начинается с крошечных примеров, чтобы вы рассуждали о правильности до перехода к масштабу.
Вы создадите SparkSession, выведете DataFrame, изучите схему и число строк и добавите запись перед повторным запуском. Подготовленная среда использует local[2]: Spark работает на одной машине с двумя потоками выполнения. Это обучает настоящим API Spark, а распределённое развёртывание остаётся отдельной темой. Нужны базовые переменные, списки и вызовы функций Python; с предварительными знаниями поможет путь Python. pandas, Hadoop, Hive, Java, установка и файлы прежних работ не требуются. Используйте терминал.
Запустите полный скрипт небольшого DataFrame
На этом шаге вы запустите полный скрипт, превращающий три записи Python в таблицу Spark. Начните в рабочем каталоге:
cd /home/labex/project
SparkSession — точка входа для создания DataFrame и доступа к API структурированных данных Spark. Построитель задаёт имя приложения, выбирает локальное выполнение и создаёт сеанс. local[2] разрешает два потока выполнения на этой машине, а не создаёт две машины кластера. setLogLevel сокращает последующие информационные журналы, чтобы учебный вывод было легче найти.
В исходном Python from pyspark.sql import SparkSession делает установленный класс доступным программе. Точка обращается к именованному члену, а скобки вызывают метод: df.show() просит DataFrame показать строки. Построитель объединяет несколько вызовов; окружающие скобки позволяют разместить выражение на нескольких строках. Каждую строку можно читать как одну настройку перед получением сеанса через getOrCreate().
Список orders содержит три кортежа — группы значений в скобках. Каждый кортеж задаёт строку в том же порядке, что объявленные столбцы. Строка схемы задаёт имя и тип каждого столбца: текстовые идентификаторы и товары, затем целое количество. createDataFrame использует записи и схему. show просит Spark вывести строки, а stop освобождает сеанс после завершения.
Сохраните полный скрипт. Символ оболочки > записывает указанный файл; маркер PYTHON в кавычках начинает here-документ, буквально сохраняющий заключённый текст Python. Копируйте до последней строки PYTHON включительно:
cat > /home/labex/project/first_dataframe.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = (
SparkSession.builder
.appName("FirstOrderTable")
.master("local[2]")
.getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")
orders = [
("s1", "notebook", 2),
("s2", "pen", 1),
("s3", "folder", 3),
]
columns = "order_id string, product string, quantity int"
df = spark.createDataFrame(orders, schema=columns)
df.show()
spark.stop()
PYTHON
Подготовленная команда bigdata-python выбирает среду Python с установленным PySpark. Запустите файл как обычную программу Python, сохраняя обычный вывод локально:
bigdata-python /home/labex/project/first_dataframe.py > /home/labex/project/output/first-run.txt
cat /home/labex/project/output/first-run.txt
Ожидается таблица со столбцами order_id, product и quantity, содержащая s1/notebook/2, s2/pen/1 и s3/folder/3. Предупреждения запуска или сообщения о ходе выполнения могут всё ещё появляться в терминале; они отделены от сохранённой таблицы. Без явной сортировки порядок строк не гарантирован, поэтому определяйте записи по ID заказов.
Ваша программа Python — драйвер: она создаёт сеанс и просит Spark выполнить работу. В кластере процессы исполнителей (executor) выполняют задачи над частями данных. Здесь выполнение происходит локально на той же машине. У вас есть работающая программа без настройки кластера.
Изучите схему и подсчитайте строки
На этом шаге вы отличите допустимое содержимое столбца от числа записей таблицы. Схема перечисляет имена столбцов, типы данных и допустимость NULL. Целочисленный столбец представляет целые числа, подходящие для расчётов количества; текст поддерживает другие операции. Nullable означает допустимость пропуска в поле, а не наличие пропуска в каждой строке.
Добавьте printSchema для изучения описания и count для подсчёта строк. Ниже полный обновлённый файл, поэтому скрытого интерактивного состояния нет:
cat > /home/labex/project/first_dataframe.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = (
SparkSession.builder
.appName("FirstOrderTable")
.master("local[2]")
.getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")
orders = [
("s1", "notebook", 2),
("s2", "pen", 1),
("s3", "folder", 3),
]
columns = "order_id string, product string, quantity int"
df = spark.createDataFrame(orders, schema=columns)
df.show()
df.printSchema()
print("Row count:", df.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/first_dataframe.py > /home/labex/project/output/schema-and-count.txt
cat /home/labex/project/output/schema-and-count.txt
Таблица должна по-прежнему содержать три записи. Под ней найдите:
root
|-- order_id: string (nullable = true)
|-- product: string (nullable = true)
|-- quantity: integer (nullable = true)
Row count: 3
printSchema описывает DataFrame. show и count — действия (actions), запрашивающие у Spark видимый результат или значение. Spark может откладывать значительную часть обработки до запроса ответа действием; в последующих работах ленивое выполнение будет объяснено через преобразования и планы. Число 3 означает три записи заказов, а не сумму количеств, которая ответила бы на другой бизнес-вопрос.
Каждый вызов запускает новый процесс Python и создаёт собственную SparkSession. Файл явно создаёт df; он не полагается на создание spark или df интерактивной оболочкой ранее.
Предположите и проверьте добавление одной записи
На этом шаге вы измените данные, сохранив форму. Предположите, что изменит добавление s4/notebook/1: число строк заказов станет четырьмя, схема останется прежней, а исходные три записи сохранятся. Новая строка — один заказ, хотя notebook уже встречается в другом заказе.
Замените скрипт этой версией, добавляющей только указанную запись:
cat > /home/labex/project/first_dataframe.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = (
SparkSession.builder
.appName("FirstOrderTable")
.master("local[2]")
.getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")
orders = [
("s1", "notebook", 2),
("s2", "pen", 1),
("s3", "folder", 3),
("s4", "notebook", 1),
]
columns = "order_id string, product string, quantity int"
df = spark.createDataFrame(orders, schema=columns)
df.show()
df.printSchema()
print("Row count:", df.count())
spark.stop()
PYTHON
Снова запустите файл в новом процессе:
bigdata-python /home/labex/project/first_dataframe.py > /home/labex/project/output/expanded-run.txt
cat /home/labex/project/output/expanded-run.txt
Найдите все четыре ID заказов и Row count: 4. Схема должна по-прежнему описывать те же три имени столбцов и типы. Повторение имени товара не группирует заказы автоматически и не удаляет дубликаты; эти операции нужно явно запросить в последующих занятиях.
Прежние first-run.txt и schema-and-count.txt остаются свидетельствами версии с тремя строками. Исполняемый файл теперь описывает версию с четырьмя строками, поэтому другой независимый запуск может воспроизвести результат. Вы завершили небольшой цикл «изменить — предположить — запустить — проверить», который поможет оценивать более крупные процессы данных, а не просто доверять успешному завершению программы.
Итоги
Вы запустили настоящий скрипт PySpark с собственной SparkSession, создали типизированный DataFrame и проверили показанные записи, схему и число строк. Добавление записи изменило число строк, сохранив схему. Далее вы перейдёте от записей внутри Python к CSV-файлу с явной схемой.
