JSON イベントを Spark DataFrame に読み込む

はじめに

記録を CSV の行で渡すシステムもあれば、名前付きフィールドを持つ JSON オブジェクトで渡すシステムもあります。今度は、小売業者が種類のラベルと注文関連の値を含むイベントファイルを提供します。処理で考えることは同じです。フィールドは何を意味し、どの記録が注文に属し、結果にどの列を出すべきでしょうか？

JSON Lines は、1行に完全な JSON オブジェクトを1つ保存します。CSV と違い、フィールド名が各オブジェクトに含まれ、オブジェクトの中に別のオブジェクトを置けます。まず平坦な記録を明示的な型で読み、次に単純な入れ子の顧客オブジェクトからフィールドを選び、最後に同じスキーマを2つ目のデータに適用します。

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、quantity、価格、日付は、CSV 注文と同じ種類の値です。ただし、在庫イベントも含まれています。有効な記録を読めたからといって、注文の結果に含めるとは限りません。以下の明示的スキーマは平坦なフィールドを宣言します。json(path) でこの形式を読み、既習の filter で event_type == order を選びます。

引用符付きのヒアドキュメントで完全なスクリプトを保存し、準備済みの 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

最初の表示には3イベントすべてが含まれるはずです。Order events の下には、j1/2/10.00/2026-10-14 と j2/1/25.00/2026-10-14 だけが期待されます。在庫記録も正常に読めましたが、種類の条件で除外されました。スキーマでは quantity は integer、価格は decimal、order_date は date になるはずです。

リーダーは、名前付き JSON フィールドを宣言した型で扱います。単純な CSV と違い、オブジェクト内のフィールド順は列の識別を決めません。それでも、意図した取り決めを定義するのはスキーマです。

入れ子の顧客オブジェクトからフィールドを選ぶ

このステップでは、構造をもう1段深く読みます。顧客イベントのファイルを確認します。

cat /home/labex/project/data/customer-events.jsonl

customer の値は、customer_id と region を含む別のオブジェクトです。Spark のスキーマでは、名前付きの子フィールドを含む struct になります。配列でも、結合する別テーブルでもありません。

内側の StructType で struct を宣言します。col("customer.region") はドット区切りのパスで子を選び、alias("region") は結果に単純な列名を付けます。顧客 ID にも同じ考え方を適用します。

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 が2つの文字列の子を持つ 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

出力を予想します。10月16日の j5 が1注文で、顧客 c2、地域 West、数量3、価格12.50です。リーダーと入れ子の選択はそのままで、入力パスだけを変えられます。完全な版を実行します。

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

予想した1行と、同じ選択後の型が期待されます。合意した構造が同じなら、新しいファイルごとに別のスキーマを考える必要はありません。

3回とも spark.read で既存のファイルを読み、スクリプトの完了とともに終了しました。将来のメッセージを購読してはいません。実行モデルまで同時に変えず、入力形式を変える方法を学べました。次は意図的に不完全にした注文データに戻り、明確なクレンジングルールを定義します。

まとめ

明示的なスカラー型で静的 JSON Lines を読み、注文イベントを選び、名前付きの列パスで単純な顧客 struct を平坦化しました。同じスキーマで2つ目のデータも正しく扱えました。次は、明確なルールの下で、欠損・重複・不正な注文記録を処理します。