Введение
Одни системы передают записи как строки CSV, другие — как объекты JSON с именованными полями. Теперь компания предоставляет файлы событий с меткой типа и значениями, связанными с заказом. Вопросы обработки знакомы: что означают поля, какие записи относятся к заказам и какие столбцы должен предоставлять результат?
JSON Lines хранит один полный объект JSON в каждой строке. В отличие от CSV, имена полей передаются с каждым объектом, а объект может содержать другой объект. Сначала вы прочитаете плоские записи с явными типами, затем выберете поля одного простого вложенного объекта клиента и наконец примените схему ко второй порции данных.
Это конечные файлы, читаемые через spark.read, поэтому обработка остаётся пакетной. Имя файла со словом events не делает приложение потоковым. Среда и учебные данные независимы; нужны прежние понятия явной схемы, select, filter и alias и базовый Python. Kafka, Hadoop, Hive, настройка потоковой обработки, массивы и сложная рекурсия не требуются. Используйте терминал.
Прочитайте плоские JSON Lines и выберите события заказов
На этом шаге вы перенесёте чтение с явной схемой с CSV на JSON. Изучите предоставленный файл:
cd /home/labex/project
cat /home/labex/project/data/flat-events.jsonl
Каждая строка — объект в фигурных скобках. Имена в двойных кавычках определяют поля; двоеточие отделяет имя от значения. Числа вроде quantity — числовые значения JSON, а order_date — строка даты в кавычках, которую интерпретирует объявленный DateType. Здесь нет строки заголовка CSV и внешнего массива.
order_id, количество, цена и дата первого объекта описывают те же виды значений, что заказ CSV. В файле также есть событие inventory; успешное чтение записи не означает, что она относится к результату заказов. Явная схема ниже объявляет плоские поля. json(path) читает этот формат, а знакомый filter выбирает event_type == order.
Сохраните полный скрипт here-документом в кавычках и запустите подготовленным интерпретатором PySpark:
cat > /home/labex/project/read_json_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("ReadJSONEvents").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
event_schema = StructType([
StructField("event_type", StringType(), True),
StructField("order_id", StringType(), True),
StructField("quantity", IntegerType(), True),
StructField("unit_price", DecimalType(10, 2), True),
StructField("order_date", DateType(), True),
])
events = (
spark.read.schema(event_schema)
.option("dateFormat", "yyyy-MM-dd")
.json("/home/labex/project/data/flat-events.jsonl")
)
events.printSchema()
events.orderBy("order_id").show(truncate=False)
orders = events.filter(col("event_type") == "order")
print("Order events:")
orders.select("order_id", "quantity", "unit_price", "order_date").orderBy("order_id").show(truncate=False)
spark.stop()
PYTHON
bigdata-python /home/labex/project/read_json_orders.py > /home/labex/project/output/flat-json-orders.txt
cat /home/labex/project/output/flat-json-orders.txt
Первый вывод должен включать все три события. Под Order events ожидаются только j1/2/10.00/2026-10-14 и j2/1/25.00/2026-10-14. Запись inventory прочитана успешно, но исключена условием типа. Схема должна показывать целое quantity, десятичный unit_price и дату order_date.
Средство чтения использует именованные поля JSON с объявленными типами; в отличие от обычной записи CSV, порядок полей внутри объекта JSON не определяет принадлежность столбцу. Схема по-прежнему задаёт нужный контракт.
Выберите поля вложенного объекта клиента
На этом шаге вы прочитаете ещё один уровень структуры. Изучите файл событий клиентов:
cat /home/labex/project/data/customer-events.jsonl
Значение customer — другой объект, содержащий customer_id и region. В схеме Spark это struct, поле с именованными дочерними полями. Это не массив и не вторая таблица для соединения.
Объявите структуру через внутренний StructType. col("customer.region") выбирает дочернее поле по пути с точкой; alias("region") задаёт выбранному результату простое имя столбца. Примените ту же идею к идентификатору клиента:
cat > /home/labex/project/read_json_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("ReadJSONEvents").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
event_schema = StructType([
StructField("event_type", StringType(), True),
StructField("order_id", StringType(), True),
StructField("customer", StructType([
StructField("customer_id", StringType(), True),
StructField("region", StringType(), True),
]), True),
StructField("quantity", IntegerType(), True),
StructField("unit_price", DecimalType(10, 2), True),
StructField("order_date", DateType(), True),
])
events = (
spark.read.schema(event_schema)
.option("dateFormat", "yyyy-MM-dd")
.json("/home/labex/project/data/customer-events.jsonl")
)
events.printSchema()
orders = events.filter(col("event_type") == "order")
selected = orders.select(
"order_id",
col("customer.customer_id").alias("customer_id"),
col("customer.region").alias("region"),
"quantity", "unit_price", "order_date",
)
print("Selected customer orders:")
selected.orderBy("order_id").show(truncate=False)
selected.printSchema()
spark.stop()
PYTHON
bigdata-python /home/labex/project/read_json_orders.py > /home/labex/project/output/nested-json-orders.txt
cat /home/labex/project/output/nested-json-orders.txt
Полная схема должна показывать customer как struct с двумя строковыми дочерними полями. Выбранная таблица должна содержать j3/c1/East/2/10.00/2026-10-15 и j4/c2/West/1/25.00/2026-10-15. Итоговая схема предоставляет customer_id и region как обычные строковые столбцы рядом с количеством, ценой и датой.
Вы выбрали значения, уже находившиеся внутри каждого события заказа. Справочник и соединение не потребовались. Первоначальное вложенное представление остаётся доступным в events, а selected даёт дальнейшим операциям более простой плоский вид.
Повторно используйте вложенную схему на другом файле
На этом шаге вы измените порцию данных, сохранив структуру. Сначала прочитайте новый файл:
cat /home/labex/project/data/new-customer-events.jsonl
Предположите вывод: один заказ j5 для c2 в West, количество 3, цена 12.50, дата 16 октября. Чтение и выбор вложенных полей могут остаться прежними; меняется только входной путь. Запустите полную версию:
cat > /home/labex/project/read_json_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("ReadJSONEvents").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
event_schema = StructType([
StructField("event_type", StringType(), True),
StructField("order_id", StringType(), True),
StructField("customer", StructType([
StructField("customer_id", StringType(), True),
StructField("region", StringType(), True),
]), True),
StructField("quantity", IntegerType(), True),
StructField("unit_price", DecimalType(10, 2), True),
StructField("order_date", DateType(), True),
])
events = (
spark.read.schema(event_schema)
.option("dateFormat", "yyyy-MM-dd")
.json("/home/labex/project/data/new-customer-events.jsonl")
)
events.printSchema()
orders = events.filter(col("event_type") == "order")
selected = orders.select(
"order_id",
col("customer.customer_id").alias("customer_id"),
col("customer.region").alias("region"),
"quantity", "unit_price", "order_date",
)
print("Selected customer orders:")
selected.orderBy("order_id").show(truncate=False)
selected.printSchema()
spark.stop()
PYTHON
bigdata-python /home/labex/project/read_json_orders.py > /home/labex/project/output/new-json-orders.txt
cat /home/labex/project/output/new-json-orders.txt
Ожидается предсказанный результат из одной строки и те же выбранные типы. Новый файл не требует изобретать новую схему, если согласованная структура не изменилась.
Все три чтения использовали spark.read над существующими файлами и закончились при завершении скриптов. Они не подписывались на будущие сообщения. Вы научились менять входной формат без одновременной смены модели выполнения. Далее вы вернётесь к намеренно несовершенным заказам и определите явные правила очистки.
Итоги
Вы прочитали статические JSON Lines с явными скалярными типами, выбрали события заказов и развернули простую структуру клиента по именованным путям столбцов. Вторая порция данных правильно использовала ту же схему. Далее вы обработаете отсутствующие, повторяющиеся и неверные записи заказов по ясным правилам очистки.
