はじめに
小売業者は注文記録を受け取り、役立つ情報にしたいと考えています。売上レポートを計算する前に、行を表し、その列を定義し、処理ツールに作業を実行させる方法が必要です。目で確認できる小さな3件の記録から始めます。
ビッグデータとは、規模、到着する速さ、処理の必要性が、手軽な単一マシンでの方法を超えるようなデータ処理を指します。Apache Spark は、クラスターの複数マシンで作業を分担できるデータ処理エンジンです。PySpark はその Python インターフェースです。DataFrame は、名前と型を持つ列で行を表し、絞り込みや集計などの操作を提供します。Spark はデータを処理し、永続的な入力と出力は別のファイルやストレージシステムが提供します。このコースでは、規模を考える前に正しさを判断できるよう、小さな例から始めます。
SparkSession を作り、DataFrame を表示し、スキーマと行数を調べ、1件加えて再実行します。準備済み環境は local[2] を使い、Spark がこの1台で2つの実行スレッドを使います。実際の Spark API を学べますが、分散構成での運用は別の内容です。Python の基本的な変数、リスト、関数呼び出しが必要です。前提知識には Python 学習パスが役立ちます。pandas、Hadoop、Hive、Java、インストール、以前の実験のファイルは不要です。ターミナルを使ってください。
小さな DataFrame の完全なスクリプトを実行する
このステップでは、Python の3記録を Spark の表にする完全なスクリプトを実行します。作業ディレクトリから始めます。
cd /home/labex/project
SparkSession は、DataFrame の作成と Spark の構造化データ API を使うための入り口です。builder でアプリケーション名とローカル実行を設定し、セッションを作ります。local[2] はこのマシンで2つの実行スレッドを許可するもので、クラスターのマシンを2台作るわけではありません。setLogLevel は、その後の情報ログを減らして、学習用の出力を見つけやすくします。
Python のソースで、from pyspark.sql import SparkSession は、インストール済みのクラスをプログラムで使えるようにします。ドットは名前付きメンバーにアクセスし、括弧はメソッドを呼び出します。df.show() は、この DataFrame の行を表示する要求です。builder は複数の呼び出しを連結し、外側の括弧で式を複数行に分けられます。getOrCreate() がセッションを得る前の各行を、1つずつの設定として読めます。
orders リストには3つのタプル、つまり括弧でまとめた値の組があります。各タプルは、宣言した列と同じ順番で1行分を指定します。スキーマ文字列は、列名と型を示します。識別子と商品はテキスト、その後は整数の数量です。createDataFrame は記録とスキーマを使い、show は Spark に行の表示を求め、stop は終了時にセッションを解放します。
完全なスクリプトを保存します。シェルの > は指定ファイルに書き込み、引用符付きの PYTHON マーカーは、囲まれた Python をそのまま保つヒアドキュメントを始めます。最後の PYTHON 行までコピーしてください。
cat > /home/labex/project/first_dataframe.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = (
SparkSession.builder
.appName("FirstOrderTable")
.master("local[2]")
.getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")
orders = [
("s1", "notebook", 2),
("s2", "pen", 1),
("s3", "folder", 3),
]
columns = "order_id string, product string, quantity int"
df = spark.createDataFrame(orders, schema=columns)
df.show()
spark.stop()
PYTHON
準備済みの bigdata-python コマンドは、PySpark が入った Python 環境を選びます。通常の Python プログラムとしてファイルを実行し、標準出力をローカルに保存します。
bigdata-python /home/labex/project/first_dataframe.py > /home/labex/project/output/first-run.txt
cat /home/labex/project/output/first-run.txt
order_id、product、quantity の列があり、s1/notebook/2、s2/pen/1、s3/folder/3 を含む表が期待されます。ターミナルには起動時の警告や進行状況も表示される場合がありますが、保存された表とは別です。明示的な並べ替えなしでは行順は保証されないので、注文 ID で記録を見つけてください。
Python プログラムはドライバーです。セッションを作り、Spark に作業を依頼します。クラスターでは、Executor プロセスがデータの各部分に対してタスクを実行します。ここでは同じマシン上でローカルに実行します。クラスターを設定せずに動くプログラムができました。
スキーマと行数を調べる
このステップでは、列が格納できる内容と、表の記録数を区別します。スキーマは、列名、データ型、NULL を許可するかを列挙します。整数列は数量計算に適した整数を表し、テキストでは使える操作が異なります。Nullable はフィールドに欠損値を許すという意味で、全行で欠損しているという意味ではありません。
printSchema でこの定義を調べ、count で行数を求めます。対話操作の隠れた状態に頼らないよう、更新後のファイル全体を示します。
cat > /home/labex/project/first_dataframe.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = (
SparkSession.builder
.appName("FirstOrderTable")
.master("local[2]")
.getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")
orders = [
("s1", "notebook", 2),
("s2", "pen", 1),
("s3", "folder", 3),
]
columns = "order_id string, product string, quantity int"
df = spark.createDataFrame(orders, schema=columns)
df.show()
df.printSchema()
print("Row count:", df.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/first_dataframe.py > /home/labex/project/output/schema-and-count.txt
cat /home/labex/project/output/schema-and-count.txt
表には引き続き3記録があるはずです。その下で次を探します。
root
|-- order_id: string (nullable = true)
|-- product: string (nullable = true)
|-- quantity: integer (nullable = true)
Row count: 3
printSchema は DataFrame の定義を表示します。show と count はアクションで、見える結果や値を Spark に求めます。Spark は、アクションが答えを要求するまで多くの処理を先送りできます。後の実験で、変換と計画を使ってこの遅延実行を説明します。件数3は注文記録3件のことで、数量の合計ではありません。数量の合計は別の業務上の疑問への答えです。
実行ごとに新しい Python プロセスが始まり、それぞれ SparkSession を作ります。ファイル内で df を明示的に作り、対話シェルが以前作った spark や df には依存しません。
記録を1つ加えた結果を予想して試す
このステップでは、データの形を保ったまま内容を変えます。s4/notebook/1 を加えると何が変わるか予想してください。注文行は4つになり、スキーマは同じで、元の3記録も残ります。notebook がほかの注文にあっても、新しい行は1件の注文です。
その記録だけを加えた次の版で、スクリプトを置き換えます。
cat > /home/labex/project/first_dataframe.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = (
SparkSession.builder
.appName("FirstOrderTable")
.master("local[2]")
.getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")
orders = [
("s1", "notebook", 2),
("s2", "pen", 1),
("s3", "folder", 3),
("s4", "notebook", 1),
]
columns = "order_id string, product string, quantity int"
df = spark.createDataFrame(orders, schema=columns)
df.show()
df.printSchema()
print("Row count:", df.count())
spark.stop()
PYTHON
新しいプロセスで、もう一度ファイルを実行します。
bigdata-python /home/labex/project/first_dataframe.py > /home/labex/project/output/expanded-run.txt
cat /home/labex/project/output/expanded-run.txt
4つの注文 ID と Row count: 4 を探します。スキーマは同じ3つの列名と型のままのはずです。商品名が繰り返されても、自動で注文がグループ化されたり、重複が削除されたりしません。これらの操作は後の実験で明示的に要求します。
以前の first-run.txt と schema-and-count.txt は、3行版の証拠として残ります。実行ファイルは4行版を定義しているので、別の独立した実行でも再現できます。変更・予想・実行・確認という小さな一巡を終えました。この習慣は、プログラムが正常終了したと信じるだけでなく、より大きなデータ処理を判断する助けになります。
まとめ
独自の SparkSession を持つ実際の PySpark スクリプトを実行し、型付き DataFrame を作って、表示された記録、スキーマ、行数を確認しました。記録を1つ加えると、スキーマを保ったまま行数が変わりました。次は、Python 内に書いた記録から、明示的なスキーマで読む CSV ファイルへ進みます。
