Введение
У компании уже есть правильный типизированный набор результатов оплаченных заказов. Теперь другому приложению нужно прочитать его позже, выбрать дату и быть уверенным, что повторная выгрузка не продублировала заказы незаметно. Сохранение данных — часть контракта обработки, а не просто последняя строка скрипта.
Вы запишете Parquet с разделением каталогов по order_date. Корень набора будет содержать подкаталоги дат, а Spark восстановит столбец даты при чтении корня. Эти каталоги хранения отличаются от вычислительных разделов выполнения; следующая работа отдельно изучает выполнение.
Здесь независимо подготовлены четыре оплаченных заказа с десятичными суммами в локальном входе Parquet. Вам следует знать чтение DataFrame, фильтры и минимальную перезапись Parquet. Каталог Hive, HDFS, предварительные знания SQL, установка и прежние результаты не требуются. Используйте терминал.
Запишите набор с разделами дат и прочитайте обратно
На этом шаге вы организуете подготовленные результаты в каталоги дат. Входной набор data/prepared-orders содержит четыре заказа с теми же типизированными количеством, десятичной ценой, датой и десятичным order_amount, что использовались ранее. Parquet двоичный; изучайте его через Spark, а не ожидайте текст CSV от cat.
Начните в рабочем каталоге:
cd /home/labex/project
partitionBy("order_date") указывает записи помещать строки в каталоги, названные по дате. mode("overwrite") заменяет существующий производный набор по заданному выходному пути. Держите его отдельно от input_path, который должен остаться нетронутым.
Сохраните и выполните полную программу:
cat > /home/labex/project/write_sales_data.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("WriteSalesData").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
input_path = "/home/labex/project/data/prepared-orders"
output_path = "/home/labex/project/output/orders-by-date"
prepared = spark.read.parquet(input_path)
prepared.write.mode("overwrite").partitionBy("order_date").parquet(output_path)
print("Main read-back:")
reloaded = spark.read.parquet(output_path)
reloaded.orderBy("order_id").show(truncate=False)
reloaded.printSchema()
print("Main rows:", reloaded.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/write_sales_data.py > /home/labex/project/output/partitioned-read-back.txt
cat /home/labex/project/output/partitioned-read-back.txt
Ожидаются четыре восстановленные записи: o1 и o2 за 14 октября, o4 и o6 за 15 октября, с суммами 20.00, 25.00, 50.00 и 30.00. Main rows должно быть 4. Проверьте, что quantity — integer, цена и сумма — decimal, order_date — date.
Столбец раздела после чтения может оказаться в другой позиции, поскольку Spark восстанавливает его из имён каталогов, а не хранит обычным полем в каждом файле раздела. Принадлежность определяется именами и значениями столбцов; используйте явный select, когда следующему потребителю нужен определённый порядок. Порядок отображения тоже требует orderBy.
Изучите каталоги и выберите одну дату
На этом шаге вы свяжете структуру каталогов с восстановленным столбцом даты. Выведите результат рекурсивно; -R включает подкаталоги дат и их файлы:
ls -R /home/labex/project/output/orders-by-date
Ожидаются каталоги order_date=2026-10-14 и order_date=2026-10-15. Имена part-файлов, их число и скрытые метаданные могут меняться; не фиксируйте один part-файл как весь набор. Значимый корень — orders-by-date.
Чтение корня позволяет Spark обнаружить значения разделов. Такое обнаружение структуры файлов не требует регистрации раздела или таблицы Hive. Следующий независимый скрипт читает корень, сравнивает столбец DateType с допустимой строкой даты ISO и отдельно сохраняет записи выбранного дня:
cat > /home/labex/project/inspect_sales_data.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
spark = SparkSession.builder.appName("InspectDateDataset").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
root = "/home/labex/project/output/orders-by-date"
df = spark.read.parquet(root)
selected = df.filter(col("order_date") == "2026-10-14")
selected.orderBy("order_id").show(truncate=False)
selected.write.mode("overwrite").parquet("/home/labex/project/output/selected-day")
spark.stop()
PYTHON
bigdata-python /home/labex/project/inspect_sales_data.py > /home/labex/project/output/selected-date.txt
cat /home/labex/project/output/selected-date.txt
Ожидаются только o1 и o2 с датами 14 октября и суммами 20.00 и 25.00. Другая дата по-прежнему существует в основном наборе. Чтение с фильтром меняет возвращённые записи; оно не удаляет невыбранные каталоги.
Вы изучили организацию хранения и фактический результат запроса. На этом крошечном примере затраченное время не доказывает преимущество формата или структуры. Цель — правильный обнаруживаемый набор для последующих читателей.
Безопасно сравните повторную перезапись с добавлением
На этом шаге вы сохраните правильный основной набор и поэкспериментируете в отдельном каталоге. Предположите количества временного набора: двукратная запись тех же четырёх строк с перезаписью каждый раз оставляет четыре строки; последующее добавление полной копии даёт восемь. Дополнительные строки — дубликаты выгрузки, а не новые заказы.
Запустите полную версию. Основной путь использует перезапись. Только write-mode-demo получает намеренное добавление:
cat > /home/labex/project/write_sales_data.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("WriteSalesData").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
input_path = "/home/labex/project/data/prepared-orders"
output_path = "/home/labex/project/output/orders-by-date"
prepared = spark.read.parquet(input_path)
prepared.write.mode("overwrite").partitionBy("order_date").parquet(output_path)
print("Main read-back:")
reloaded = spark.read.parquet(output_path)
reloaded.orderBy("order_id").show(truncate=False)
reloaded.printSchema()
print("Main rows:", reloaded.count())
practice_path = "/home/labex/project/output/write-mode-demo"
prepared.write.mode("overwrite").partitionBy("order_date").parquet(practice_path)
print("First overwrite rows:", spark.read.parquet(practice_path).count())
prepared.write.mode("overwrite").partitionBy("order_date").parquet(practice_path)
print("Second overwrite rows:", spark.read.parquet(practice_path).count())
prepared.write.mode("append").partitionBy("order_date").parquet(practice_path)
print("After append rows:", spark.read.parquet(practice_path).count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/write_sales_data.py > /home/labex/project/output/write-mode-comparison.txt
cat /home/labex/project/output/write-mode-comparison.txt
Найдите Main rows: 4, First overwrite rows: 4, Second overwrite rows: 4 и After append rows: 8. Временный набор намеренно содержит две копии каждого заказа, показывая, почему добавление полного снимка не подходит основной выгрузке.
Снова запустите сохранённую программу для проверки повторяемости всего процесса:
bigdata-python /home/labex/project/write_sales_data.py > /home/labex/project/output/write-mode-comparison.txt
cat /home/labex/project/output/write-mode-comparison.txt
Должны появиться те же количества. Каждый полный запуск перестраивает основной набор и сбрасывает временный эксперимент перед намеренным добавлением, поэтому тот не растёт до двенадцати или шестнадцати строк. Добавление — отдельная политика для действительно новых непересекающихся данных; перезапись — выбранная политика полного производного снимка.
Первоначальный набор prepared-orders остаётся доступен. Каталоги дат описывают сохранённые значения, а вычислительные разделы Spark — деление работы задания во время выполнения. Далее вы свяжете второе значение слова «раздел» с планами и настоящим выполнением.
Итоги
Вы записали Parquet с разделами по датам, восстановили типизированные записи из корня и выбрали одну дату без удаления других. Повторная перезапись сохранила основной снимок, а временное добавление показало дубликаты. Далее планы выполнения и реальные события объяснят вычислительные разделы, задания, этапы и задачи.
