Введение
Теперь записи заказов компании приходят в CSV-файле, а не записываются прямо в список Python. Файл хранит текст вроде 2 и 10.00; вычислению нужно знать, означают ли значения текст, целое количество или десятичную цену. Чтение правильных символов — лишь часть правильного чтения данных.
Вы сравните стандартное строковое чтение Spark, автоматический вывод типов и явную схему. Затем примените ту же схему ко второму файлу с тем же порядком полей. Здесь данные намеренно корректны, чтобы вы сосредоточились на договорённости между столбцами и значениями. Одна из последующих работ явно обработает отсутствующие и неверные входные данные.
Среда предоставляет PySpark и два независимых локальных CSV-примера. Вам следует понимать SparkSession, вывод DataFrame, схему и число строк из первой работы, а также базовый Python. pandas, Hadoop, Hive, установка и прежнее состояние VM не требуются. Используйте терминал.
Прочитайте заголовок и изучите строковые столбцы
На этом шаге вы свяжете позиции полей CSV с таблицей Spark. Начните с просмотра фактического файла:
cd /home/labex/project
cat /home/labex/project/data/orders.csv
Первая строка называет семь столбцов; следующие шесть — заказы. Каждая строка представляет заказ одного вида товара. Файл включает статусы paid, pending и refunded; здесь вы читаете все без применения правила продаж.
spark.read создаёт средство чтения DataFrame. option("header", True) сообщает ему считать первую строку CSV именами столбцов, а не записью заказа. csv(input_path) указывает локальный файл. Без вывода типов и предоставленной схемы средство чтения CSV создаёт строковые столбцы.
Сохраните и запустите полную программу. Here-документ PYTHON в кавычках сохраняет заключённый исходный текст, а > записывает локальный файл. bigdata-python использует подготовленную среду PySpark:
cat > /home/labex/project/read_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("ReadOrderCSV").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
input_path = "/home/labex/project/data/orders.csv"
df = spark.read.option("header", True).csv(input_path)
df.printSchema()
df.orderBy("order_id").show(truncate=False)
print("Rows:", df.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/read_orders.py > /home/labex/project/output/string-orders.txt
cat /home/labex/project/output/string-orders.txt
Ожидаются шесть строк и семь строковых столбцов, включая quantity и unit_price. Показанное 2 может выглядеть числом, но printSchema определяет настоящий строковый тип. orderBy("order_id") сортирует вывод по идентификатору для предсказуемого просмотра; сортировка отделена от назначения типов.
header=True не позволил заголовку стать седьмой записью. Но он не сообщил Spark, что цены должны быть точными десятичными числами, а даты — поддерживать операции дат.
Сравните вывод типов с явным контрактом данных
На этом шаге вы объявите смысл каждого поля CSV. inferSchema=True просит Spark изучить значения и вывести типы. Это удобно, но результат зависит от наблюдаемых данных; это не долговременная договорённость о ценах и будущих файлах. В этом примере вывод определяет unit_price как double — число с плавающей точкой.
Для нужного контракта используйте явную схему. StructType — упорядоченный набор полей. Каждый StructField содержит имя, тип и True для разрешения пропусков. StringType хранит идентификаторы и текст статуса; IntegerType — количества; DecimalType(10, 2) — точные цены с десятью цифрами всего и двумя дробными; DateType — календарные даты. Допустимость NULL разрешает пропуски, но не делает отсутствующую бизнес-информацию автоматически приемлемой.
Замените программу этой версией. Она выводит описание выведенных типов для сравнения, затем читает тот же файл с явной схемой:
cat > /home/labex/project/read_orders.py <<'PYTHON'
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("ReadOrderCSV").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),
])
input_path = "/home/labex/project/data/orders.csv"
inferred = spark.read.option("header", True).option("inferSchema", True).csv(input_path)
print("Inferred schema:", inferred.schema.simpleString())
df = (
spark.read
.option("header", True)
.option("dateFormat", "yyyy-MM-dd")
.schema(order_schema)
.csv(input_path)
)
print("Explicit schema:")
df.printSchema()
df.orderBy("order_id").show(truncate=False)
print("Rows:", df.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/read_orders.py > /home/labex/project/output/typed-orders.txt
cat /home/labex/project/output/typed-orders.txt
Найдите явную схему под сводкой выведенных типов. quantity должен быть integer, unit_price — decimal(10,2), order_date — date. Все шесть записей должны остаться. Явный параметр dateFormat описывает входные даты как год-месяц-день, что соответствует значениям вроде 2026-10-14.
Порядок полей схемы совпадает с порядком полей этого CSV. Предоставленная схема не является общим указанием переставлять любой входящий файл по именам заголовка. Для этой порции с известным порядком контракт ясно задаёт нужные столбцы и типы. Схема также не заменяет политику качества данных; позже вы научитесь безопасно обрабатывать неверные значения.
Примените ту же схему к новой порции данных
На этом шаге вы проверите контракт на втором корректном файле. Сначала изучите его:
cat /home/labex/project/data/new-orders.csv
В нём те же семь полей в том же порядке, но два новых заказа. Предположите изменения: две бизнес-строки вместо шести, а количество, цена и дата сохранят объявленные типы.
Следующая полная версия сохраняет явную схему и меняет входной путь. Сравнение вывода типов для такого обычного чтения больше не нужно:
cat > /home/labex/project/read_orders.py <<'PYTHON'
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("ReadOrderCSV").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),
])
input_path = "/home/labex/project/data/new-orders.csv"
df = (
spark.read
.option("header", True)
.option("dateFormat", "yyyy-MM-dd")
.schema(order_schema)
.csv(input_path)
)
print("Explicit schema:")
df.printSchema()
df.orderBy("order_id").show(truncate=False)
print("Rows:", df.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/read_orders.py > /home/labex/project/output/new-typed-orders.txt
cat /home/labex/project/output/new-typed-orders.txt
Ожидаются o7 с количеством 2, ценой 12.50, статусом paid и датой 2026-10-16; o8 с количеством 3, ценой 4.00, статусом pending и датой 2026-10-17. Число строк должно быть 2. Проверьте те же типы integer, decimal и date в printSchema.
Первоначальный файл из шести заказов не изменён. Повторно используемая схема позволяет новой порции войти в ту же обработку без переопределения смысла количества и цены. Далее вы используете типизированные столбцы для выбора заказов и расчёта суммы каждого заказа.
Итоги
Вы правильно прочитали заголовок CSV, сравнили строковое чтение и вывод типов и объявили явный контракт идентификаторов, количеств, десятичных цен и дат. Та же схема правильно интерпретировала вторую порцию данных. Далее выражения типизированных столбцов позволят фильтровать строки и рассчитывать суммы заказов.
