明示的なスキーマで Spark の CSV を読む

はじめに

小売業者の注文記録が、Python リストに直接書かれる代わりに CSV ファイルで届くようになりました。ファイルには 2 や 10.00 というテキストが保存されていますが、計算するには、それが文字列、整数の数量、十進数の価格のどれなのかを知る必要があります。正しい文字を読むだけでは、データを正しく読んだことにはなりません。

Spark の既定の文字列読み込み、自動的な型推論、明示的なスキーマを比較します。その後、同じフィールド順の2つ目のファイルにも同じスキーマを適用します。ここでは、列と値の取り決めに集中できるよう、業務データは意図的に正しいものを使います。欠損や不正な入力は、後の実験で明示的に扱います。

環境には PySpark と独立した2つのローカル CSV が用意されています。Python の基礎と、最初の実験の SparkSession、DataFrame の表示、スキーマ、行数を理解していることを想定します。pandas、Hadoop、Hive、インストール、以前の VM の状態は不要です。ターミナルを使用してください。

ヘッダーを読み、文字列の列を観察する

このステップでは、CSV のフィールド位置と、Spark が作る表を結び付けます。実際のファイルの確認から始めます。

cd /home/labex/project

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

最初の行は7つの列名で、次の6行は注文です。各行は1種類の商品を含む1注文を表します。paid、pending、refunded の状態があり、この実験では売上ルールを適用せず全部を読みます。

spark.read は DataFrame リーダーを作ります。option("header", True) は、最初の CSV 行を注文記録ではなく列名として扱う指定です。csv(input_path) はローカルファイルを指定します。型推論もスキーマ指定もない場合、CSV リーダーは文字列の列を作ります。

完全なプログラムを保存して実行します。引用符付きの 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

6行と、quantity や unit_price も含む7つの文字列列が期待されます。表示された 2 は数値に見えても、printSchema が実際には string 型だと示します。orderBy("order_id") は、確認しやすいよう表示行を ID で並べ替えます。並べ替えと型の割り当ては別です。

header=True によって、ヘッダーが7件目の記録になるのを防ぎました。しかし、価格を正確な十進数にしたり、日付操作を可能にしたりする指示にはなりません。

型推論と明示的なデータの取り決めを比較する

このステップでは、各 CSV フィールドの意味を宣言します。inferSchema=True は Spark に値を調べさせて型を推論します。便利な場合もありますが、結果は観察したデータに依存し、価格や将来のファイルについての永続的な取り決めにはなりません。このサンプルでは unit_price は浮動小数点数 double と推論されます。

意図した取り決めには明示的なスキーマを使います。StructType は順序のあるフィールドの集合です。各 StructField には、名前、型、欠損値を許す True が入ります。StringType は ID と状態の文字列、IntegerType は数量、DecimalType(10, 2) は全体10桁・小数点以下2桁の正確な価格、DateType はカレンダーの日付を格納します。Nullable は欠損を許しますが、業務上の情報の欠損を自動的に認めるわけではありません。

プログラムを次の版に置き換えます。比較用に推論した型の説明を表示した後、同じファイルを明示的なスキーマで読みます。

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 のはずです。6記録はすべて残ります。明示した dateFormat は、2026-10-14 のような値に合わせ、入力日付を年・月・日で表すと指定します。

スキーマのフィールド順は、この CSV のフィールド順に合っています。指定したスキーマは、あらゆる入力ファイルをヘッダー名で並べ直す一般的な指示ではありません。順番が既知の今回のデータでは、取り決めによって目的の列と型が明確になります。また、スキーマはデータ品質の方針の代わりではありません。不正値を安全に扱う方法は後で学びます。

同じスキーマを新しいファイルに適用する

このステップでは、2つ目の正しいファイルで取り決めを試します。最初に確認してください。

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

同じ7フィールドが同じ順で並び、新しい2注文が入っています。変化を予想しましょう。業務行は6から2になり、数量、価格、日付の型は宣言どおりのままです。

次の完全な版は、明示的なスキーマを保ち、入力パスを変えています。通常の読み取りなので、型推論との比較はもう必要ありません。

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です。printSchema で同じ整数・十進数・日付型も確かめてください。

元の6注文のファイルは変わりません。再利用できるスキーマにより、数量や価格の意味を再定義せず、新しいデータを同じ処理ロジックに入れられます。次は、型付きの列で注文を選び、各注文の金額を計算します。

まとめ

CSV ヘッダーを正しく読み、文字列の読み込みと型推論を比べ、ID、数量、十進数の価格、日付の取り決めを明示しました。同じスキーマで2つ目のファイルも正しく解釈できました。次は、型付きの列表現で行を絞り込み、注文金額を計算します。