Выбирайте, фильтруйте и преобразуйте заказы Spark

Введение

Правильно типизированная таблица заказов — отправная точка, но ещё не ответ. Коллеге могут понадобиться лишь несколько столбцов, только оплаченные заказы и сумма, рассчитанная по количеству и цене единицы. Вы выразите эти решения операциями DataFrame Spark, сохранив исходные записи доступными.

Выражение столбца описывает вычисление или условие, которое Spark применит к строкам. Вы выберете поля, объедините условия, добавите десятичную сумму и дадите выходному столбцу понятное имя. Эти операции создают новые DataFrame; действия вроде show запрашивают фактические результаты. Это различие помогает объяснить ленивое выполнение Spark без изучения внутреннего планирования.

Работа независимо предоставляет шесть чистых заказов. Вам следует понимать явную схему CSV из предыдущей работы и базовый Python. Каждая запускаемая версия включает полное чтение и SparkSession, поэтому прежний процесс или файл не нужен. Используйте терминал и подготовленную среду bigdata-python.

Выберите столбцы, нужные для вопроса

На этом шаге вы сократите столбцы таблицы, сохранив все шесть заказов. Начните с источника:

cd /home/labex/project

cat /home/labex/project/data/orders.csv

select выбирает столбцы нового DataFrame и их порядок. Он не фильтрует строки. Полный скрипт ниже повторяет знакомое явное чтение, затем выбирает идентификатор заказа, количество, цену и статус. Дата и поля клиента остаются в df, хотя selected их не предоставляет.

Сохраните скрипт here-документом в кавычках и запустите. > записывает файл или заменяет предыдущую версию; последний маркер PYTHON завершает исходный текст:

cat > /home/labex/project/transform_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType

spark = SparkSession.builder.appName("TransformOrders").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
order_schema = StructType([
    StructField("order_id", StringType(), True),
    StructField("customer_id", StringType(), True),
    StructField("product_id", StringType(), True),
    StructField("quantity", IntegerType(), True),
    StructField("unit_price", DecimalType(10, 2), True),
    StructField("order_status", StringType(), True),
    StructField("order_date", DateType(), True),
])
df = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")

selected = df.select("order_id", "quantity", "unit_price", "order_status")
selected.orderBy("order_id").show(truncate=False)

spark.stop()
PYTHON

bigdata-python /home/labex/project/transform_orders.py > /home/labex/project/output/selected-orders.txt

cat /home/labex/project/output/selected-orders.txt

Ожидаются все шесть ID заказов с четырьмя показанными столбцами. orderBy делает вывод предсказуемым, а show запускает работу, необходимую для получения строк. Это выбор столбцов, иногда называемый проекцией. Исходный файл не изменён, заказы не удалены.

Объедините условия выбора строк

На этом шаге вы оставите оплаченные заказы с количеством не менее двух. col("order_status") ссылается на столбец Spark; сравнение с == "paid" строит условие Spark, а не одно значение Python True или False. >= строит условие количества.

Используйте & для объединения условий столбцов Spark со смыслом AND. Поместите каждое сравнение в скобки, поскольку приоритет операторов Python имеет значение. Обычный Python and не является оператором объединения объектов Spark Column.

Предположите подходящие ID: o1, o4 и o6 удовлетворяют обоим условиям. o2 оплачен, но его количество равно единице. Замените и запустите полный скрипт:

cat > /home/labex/project/transform_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType

spark = SparkSession.builder.appName("TransformOrders").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
order_schema = StructType([
    StructField("order_id", StringType(), True),
    StructField("customer_id", StringType(), True),
    StructField("product_id", StringType(), True),
    StructField("quantity", IntegerType(), True),
    StructField("unit_price", DecimalType(10, 2), True),
    StructField("order_status", StringType(), True),
    StructField("order_date", DateType(), True),
])
df = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")

eligible = df.filter((col("order_status") == "paid") & (col("quantity") >= 2))
eligible.select("order_id", "quantity", "unit_price", "order_status").orderBy("order_id").show(truncate=False)

spark.stop()
PYTHON

bigdata-python /home/labex/project/transform_orders.py > /home/labex/project/output/eligible-orders.txt

cat /home/labex/project/output/eligible-orders.txt

Результат должен содержать только эти три ID, каждый с количеством 2. filter меняет выбранные строки; select управляет показанными столбцами. Можно объединить обе операции, пока df продолжает описывать полный источник из шести строк.

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

Рассчитайте и назовите десятичные суммы заказов

На этом шаге вы вычислите сумму каждого оплаченного заказа как количество, умноженное на десятичную цену единицы. withColumn добавляет именованный столбец в новый DataFrame или заменяет его, если имя уже существует. Здесь order_amount рассчитывается встроенным выражением умножения Spark.

Следующий select сохраняет ID заказа и даёт вычисленному столбцу отображаемое имя amount через alias. Псевдоним меняет имя столбца результата, а не редактирует заголовок CSV. Все четыре оплаченных заказа относятся к этому расчёту; строки pending и refunded исключены, а возвраты не считаются отрицательной выручкой в упрощённом правиле.

Запустите полную версию:

cat > /home/labex/project/transform_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType

spark = SparkSession.builder.appName("TransformOrders").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
order_schema = StructType([
    StructField("order_id", StringType(), True),
    StructField("customer_id", StringType(), True),
    StructField("product_id", StringType(), True),
    StructField("quantity", IntegerType(), True),
    StructField("unit_price", DecimalType(10, 2), True),
    StructField("order_status", StringType(), True),
    StructField("order_date", DateType(), True),
])
df = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")

paid_result = (
    df.filter(col("order_status") == "paid")
    .withColumn("order_amount", col("quantity") * col("unit_price"))
    .select("order_id", col("order_amount").alias("amount"))
)
print("Paid orders:")
paid_result.orderBy("order_id").show(truncate=False)
paid_result.printSchema()

spark.stop()
PYTHON

bigdata-python /home/labex/project/transform_orders.py > /home/labex/project/output/paid-amounts.txt

cat /home/labex/project/output/paid-amounts.txt

Ожидаются o1/20.00, o2/25.00, o4/50.00 и o6/30.00. Могут появиться дополнительные конечные нули. Схема должна определять amount как десятичный тип. Spark может расширять точность decimal при умножении; главное — сохранение десятичной арифметики без преобразования цен в двоичную плавающую точку.

filter, withColumn, select и orderBy описывают преобразования (transformations) и возвращают новые DataFrame. Определение цепочки не материализует сразу каждую промежуточную таблицу. show — действие, запрашивающее выполнение необходимой работы Spark и показ ответа. printSchema описывает структуру столбцов. Позже вы изучите фактический план выполнения такой цепочки.

Измените один статус и предположите эффект

На этом шаге вы сохраните результат оплаченных заказов и добавите отдельное учебное представление ожидающих. Предположите вывод перед запуском: только o3 имеет статус pending, количество один и цену 10.00, поэтому сумма равна 10.00.

Новый pending_result использует ту же цепочку выражений с изменением только значения статуса. Он начинается с df, а не с уже отфильтрованного paid_result, поэтому ожидающие записи всё ещё доступны:

cat > /home/labex/project/transform_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType

spark = SparkSession.builder.appName("TransformOrders").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
order_schema = StructType([
    StructField("order_id", StringType(), True),
    StructField("customer_id", StringType(), True),
    StructField("product_id", StringType(), True),
    StructField("quantity", IntegerType(), True),
    StructField("unit_price", DecimalType(10, 2), True),
    StructField("order_status", StringType(), True),
    StructField("order_date", DateType(), True),
])
df = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")

paid_result = (
    df.filter(col("order_status") == "paid")
    .withColumn("order_amount", col("quantity") * col("unit_price"))
    .select("order_id", col("order_amount").alias("amount"))
)
print("Paid orders:")
paid_result.orderBy("order_id").show(truncate=False)
paid_result.printSchema()

pending_result = (
    df.filter(col("order_status") == "pending")
    .withColumn("order_amount", col("quantity") * col("unit_price"))
    .select("order_id", col("order_amount").alias("amount"))
)
print("Pending practice:")
pending_result.orderBy("order_id").show(truncate=False)

spark.stop()
PYTHON

bigdata-python /home/labex/project/transform_orders.py > /home/labex/project/output/status-comparison.txt

cat /home/labex/project/output/status-comparison.txt

Под Paid orders должны остаться те же четыре оплаченные суммы. Под Pending practice ожидается только o3/10.00. Фильтрация оплаченного представления по pending не вернула бы строк, поскольку эти записи уже исключены; важно начинать с подходящего источника.

Обе цепочки вычисляются встроенными выражениями Spark. Вы не собирали общий набор данных в список Python для прохода циклом. Это сохраняет описание обработки подходящим движку выполнения Spark. Итоговый скрипт содержит расчёт оплаченных заказов и небольшое учебное сравнение, а первоначальный CSV по-прежнему содержит все шесть заказов.

Итоги

Вы выбрали столбцы, объединили условия строк выражениями Spark, рассчитали десятичные суммы заказов и назвали выходные столбцы. Сравнение оплаченного и ожидающего представлений показало, почему преобразования должны начинаться с подходящего источника и как действия запрашивают результаты. Далее вы примените знакомые идеи чтения и выбора к JSON.