Введение
Скрипт может завершиться успешно и всё равно передать плохие записи в отчёт. Сегодняшняя порция продаж содержит повтор заказа, отсутствующего клиента, неверный числовой текст, неположительное количество, отрицательную цену и неправильную дату. Автоматическое заполнение этих проблем нулём выдумало бы бизнес-информацию.
Вы разделите три решения: прочитать текст без потери исходных свидетельств, безопасно преобразовать нужные числовые поля и даты и оставить только строки, отвечающие указанному правилу. Затем одинаковые допустимые строки будут избавлены от дубликатов. Наконец, вы сохраните и прочитаете чистый набор Parquet. Parquet — типизированный двоичный столбцовый формат файлов; здесь изучается только минимальная запись и чтение, а хранение с разделами оставлено на потом.
Подготовленная среда предоставляет две независимые порции данных. Вам следует знать чтение CSV, выражения столбцов Spark, фильтрацию и десятичные цены. Используемое безопасное преобразование поддерживается подготовленной версией Spark; менять настройки обработки ошибок не нужно. Используйте терминал. Предыдущие результаты, Hadoop, Hive и установка служб не требуются.
Сохраните порцию данных как промежуточные строки
На этом шаге вы увидите дефекты перед преобразованием значений. Изучите локальный источник:
cd /home/labex/project
cat /home/labex/project/data/dirty-orders.csv
Здесь четырнадцать бизнес-строк. Промежуточный (staging) DataFrame сохраняет входные поля как строки, позволяя видеть two, unknown и not-a-date без предположения, что это допустимые числа и даты. Стандартная обработка CSV делает пустое поле клиента NULL; NULL представляет пропуск, а не текст «NULL».
Сохраните и запустите полную промежуточную программу. Как и прежде, here-документ PYTHON в кавычках буквально записывает исходник. show(30, truncate=False) разрешает вывод до тридцати записей и сохраняет читаемый текст полей:
cat > /home/labex/project/clean_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, trim, expr
spark = SparkSession.builder.appName("CleanOrderDelivery").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
staging = spark.read.option("header", True).csv("/home/labex/project/data/dirty-orders.csv")
staging.printSchema()
staging.orderBy("order_id").show(30, truncate=False)
print("Staging rows:", staging.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/clean_orders.py > /home/labex/project/output/staging-orders.txt
cat /home/labex/project/output/staging-orders.txt
Ожидаются четырнадцать показанных записей и Staging rows: 14. Все столбцы должны быть строковыми. Найдите количество two у b2 и цену unknown у b4; это полезные свидетельства, а не ошибки, скрытые угаданным значением по умолчанию. o2 встречается дважды, а статус o1 окружён пробелами.
Исходный файл в упражнении доступен только для чтения. Вы создадите производные DataFrame и выходные каталоги, оставив первоначальные данные доступными для проверки.
Безопасно преобразуйте значения без выдуманных замен
На этом шаге вы нормализуете два текстовых поля, которым нужно убрать крайние пробелы, и попробуете преобразовать числа и даты. trim удаляет начальные и конечные пробелы. Она не угадывает отсутствующего клиента и не меняет статус оплаты заказа.
expr позволяет передать встроенное выражение Spark текстом. try_cast(quantity AS INT) пытается выполнить целочисленное преобразование и выдаёт NULL, если значение преобразовать нельзя. Аналогичные выражения запрашивают DECIMAL(10,2) для цен и DATE для дат. Обычные неудачные преобразования могут вызывать ошибки при поведении ANSI Spark; явное безопасное преобразование не зависит от глобальной настройки.
Изменение схемы сохраняет строки, чтобы проблемы разбора оставались наблюдаемыми. Запустите эту версию:
cat > /home/labex/project/clean_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, trim, expr
spark = SparkSession.builder.appName("CleanOrderDelivery").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
staging = spark.read.option("header", True).csv("/home/labex/project/data/dirty-orders.csv")
parsed = (
staging
.withColumn("customer_id", trim(col("customer_id")))
.withColumn("order_status", trim(col("order_status")))
.withColumn("quantity", expr("try_cast(quantity AS INT)"))
.withColumn("unit_price", expr("try_cast(unit_price AS DECIMAL(10,2))"))
.withColumn("order_date", expr("try_cast(order_date AS DATE)"))
)
parsed.printSchema()
parsed.orderBy("order_id").show(30, truncate=False)
print("Parsed rows:", parsed.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/clean_orders.py > /home/labex/project/output/parsed-orders.txt
cat /home/labex/project/output/parsed-orders.txt
Ожидается Parsed rows: 14. Количество b2, цена b4 и дата b6 теперь должны быть NULL. Нулевое количество b3 и отрицательная цена b5 синтаксически допустимы как числа, но всё ещё нарушают бизнес-правила, которые вы примените далее. Статус o1 теперь должен быть paid без крайних пробелов.
Успешное преобразование не равнозначно допустимому заказу. Разделение этих этапов позволяет объяснить, не прошла ли строка разбор или бизнес-ограничение. Ни одна пропущенная цена не заменена нулём; настоящий заказ o7 с нулевой ценой остаётся отличим от неизвестной цены.
Примените явные правила допустимости и дубликатов
На этом шаге вы оставите допустимые бизнес-записи и удалите только одинаковые дубликаты. Используйте правила:
- customer_id присутствует и непуст после удаления пробелов;
- quantity успешно преобразовано и больше нуля;
- unit_price успешно преобразовано и не меньше нуля;
- order_date успешно преобразовано;
- строки, одинаковые по всем семи нормализованным полям, сохраняются один раз.
Остальные поля идентификаторов примера уже правильно сформированы. Конфликтующих версий одного ID заказа нет, поэтому упражнение не выбирает победителя среди меняющихся статусов. Paid, pending и refunded остаются, если в остальном допустимы; очистка порции данных отличается от выбора оплаченной выручки для отчёта.
isNotNull явно проверяет наличие. Сравнение с NULL не становится истинным совпадением, а filter сохраняет только истинные условия. Объединение через & делает политику видимой. dropDuplicates() без подмножества удаляет одинаковые полные строки; выбор только order_id был бы другой политикой, способной отбросить конфликтующую информацию.
Запустите эту версию:
cat > /home/labex/project/clean_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, trim, expr
spark = SparkSession.builder.appName("CleanOrderDelivery").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
staging = spark.read.option("header", True).csv("/home/labex/project/data/dirty-orders.csv")
parsed = (
staging
.withColumn("customer_id", trim(col("customer_id")))
.withColumn("order_status", trim(col("order_status")))
.withColumn("quantity", expr("try_cast(quantity AS INT)"))
.withColumn("unit_price", expr("try_cast(unit_price AS DECIMAL(10,2))"))
.withColumn("order_date", expr("try_cast(order_date AS DATE)"))
)
valid = parsed.filter(
col("customer_id").isNotNull()
& (col("customer_id") != "")
& col("quantity").isNotNull() & (col("quantity") > 0)
& col("unit_price").isNotNull() & (col("unit_price") >= 0)
& col("order_date").isNotNull()
)
clean = valid.dropDuplicates()
print("Clean rows:")
clean.orderBy("order_id").show(30, truncate=False)
print("Staging count:", staging.count())
print("Valid count before deduplication:", valid.count())
print("Clean count:", clean.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/clean_orders.py > /home/labex/project/output/cleaning-result.txt
cat /home/labex/project/output/cleaning-result.txt
Ожидаются Staging count: 14, Valid count before deduplication: 8 и Clean count: 7. b1–b6 исключены по правилам, а один повтор o2 удалён. Семь сохранённых ID — от o1 до o7.
Настоящая нулевая цена o7 допустима по правилу неотрицательной цены. o3 всё ещё pending, а o5 refunded: ни один из статусов не делает исходную запись неверной. Полученный набор чист по указанной политике; это не утверждение, что решены все возможные аномалии реального мира.
Сохраните, прочитайте и проверьте дополнительный дефект
На этом шаге вы сделаете чистый набор пригодным для повторного использования и проверите ту же политику на другой порции. write.mode("overwrite").parquet(path) записывает набор Parquet в каталог, заменяя прежнее производное содержимое. Держите каталог отдельно от исходных данных. Spark может создавать несколько part-файлов и маркер завершения; не полагайтесь на конкретное имя или число файлов.
spark.read.parquet(path) читает сохранённый набор и его типы. Запрос к прочитанному DataFrame проверяет настоящий результат, а не просто выводит план до записи. Полная программа ниже помещает уже изученную логику в функцию Python, чтобы обе порции использовали одно правило очистки.
def clean_orders(input_path): определяет функцию и называет входной параметр. Строки с отступом — её тело; они выполняются при вызове. Первый вызов передаёт исходное имя файла, второй — имя дополнительной порции. Внутри каждого вызова input_path ссылается на переданное значение. return clean возвращает полученный DataFrame вызывающему коду; файлы он не сохраняет. Строки записи без отступа выполняют это отдельное действие. Сохраняйте отступы при копировании полной программы.
Дополнительная порция содержит исходные четырнадцать строк, строку с отрицательным количеством и ещё одну точную копию допустимого заказа с нулевой ценой. Предположите чистый результат: добавления не должны изменить семь сохранённых записей. Изучите добавленные строки:
tail -n 2 /home/labex/project/data/dirty-orders-extra.csv
Запустите полную программу сохранения и чтения обратно:
cat > /home/labex/project/clean_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, trim, expr
spark = SparkSession.builder.appName("CleanOrderDelivery").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
def clean_orders(input_path):
staging = spark.read.option("header", True).csv(input_path)
parsed = (
staging
.withColumn("customer_id", trim(col("customer_id")))
.withColumn("order_status", trim(col("order_status")))
.withColumn("quantity", expr("try_cast(quantity AS INT)"))
.withColumn("unit_price", expr("try_cast(unit_price AS DECIMAL(10,2))"))
.withColumn("order_date", expr("try_cast(order_date AS DATE)"))
)
valid = parsed.filter(
col("customer_id").isNotNull()
& (col("customer_id") != "")
& col("quantity").isNotNull() & (col("quantity") > 0)
& col("unit_price").isNotNull() & (col("unit_price") >= 0)
& col("order_date").isNotNull()
)
clean = valid.dropDuplicates()
return clean
clean = clean_orders("/home/labex/project/data/dirty-orders.csv")
clean.write.mode("overwrite").parquet("/home/labex/project/output/clean-orders")
reloaded = spark.read.parquet("/home/labex/project/output/clean-orders")
print("Read-back clean rows:")
reloaded.orderBy("order_id").show(30, truncate=False)
reloaded.printSchema()
print("Read-back count:", reloaded.count())
practice = clean_orders("/home/labex/project/data/dirty-orders-extra.csv")
practice.write.mode("overwrite").parquet("/home/labex/project/output/practice-clean-orders")
practice_reloaded = spark.read.parquet("/home/labex/project/output/practice-clean-orders")
print("Extra-delivery clean rows:")
practice_reloaded.orderBy("order_id").show(30, truncate=False)
print("Extra-delivery count:", practice_reloaded.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/clean_orders.py > /home/labex/project/output/clean-parquet-check.txt
cat /home/labex/project/output/clean-parquet-check.txt
Обе прочитанные таблицы должны содержать o1–o7 ровно по одному разу, оба количества — 7. Основной выходной каталог — output/clean-orders; practice-clean-orders — отдельный производный эксперимент. Прочитанная схема должна сохранять целое quantity, десятичный unit_price и дату order_date.
Это минимальная запись и чтение Parquet, а не сравнение скорости и не упражнение настройки хранения. Теперь у вас есть небольшой чистый набор, который прочитает другой процесс Spark, и свидетельство, что новая неверная строка и дубликат не меняют результат незаметно.
Итоги
Вы сохранили исходный текст, безопасным преобразованием выявили неверные значения, применили явную политику пропусков и диапазонов и удалили одинаковые строки-дубликаты. Вы сохранили допустимые нулевые цены и неоплаченные записи, затем проверили настоящий типизированный результат Parquet и вторую порцию данных. Следующее задание позволит самостоятельно применить очистку и расчёт сумм.
