スポンサーリンク

PySparkのDataframeを3手順で結合(join)する【ソースコードあり】

★悩み★
・PySparkのDataframeで結合(join)はどうすればいいのだろうか。
・結合(join)したPySparkのDataframeから特定列の抽出手順を知りたい。


こういった「悩み」に答えます。

★本記事の内容★
① PySparkのDataframeを3手順で結合(join)する手順をご紹介
② 結合(join)したPySparkのDataframeから特定列を抽出する手順をご紹介


これからご紹介する「PySparkのDataframeを3手順で結合(join)する手順」を実践したことで、筆者は30分以内でPySparkのDataframeを結合できました。

記事の前半では「PySparkのDataframeを結合(join)する手順」をソースコードを交えながら紹介します。
記事の後半では「結合(join)したPySparkのDataframeから特定列を抽出する手順」をソースコードを交えながら紹介します。

この記事を読み終えることで、「PySparkのDataframeで結合(join)する手順」を把握できるだけでなく、「PySparkの基本的な使い方」も把握した状態になります。

Apache Sparkをインストールされていない方は、以下の記事をご覧ください。

PySparkのDataframeを3手順で結合(join)する手順

「PySparkのDataframeを3手順で結合(join)する手順」についてご紹介します。

★PySparkのDataframeを結合(join)する手順の流れ★
手順1:PySparkでデータをDataframeとして読み込み
手順2:PySparkのDataframeを結合(join)
手順3:PySparkでDataframeを結合(join)するプログラムを実行


上記の流れで、PySparkのDataframeを結合(join)することができます。

以降で、「PySparkのDataframeを結合(join)する手順の流れ」の各手順に関して説明します。

今回は、以下で紹介したデータを元に、「PySparkのDataframeを作成し結合(join)する手順」を説明します。


手順1:PySparkでデータをDataframeとして読み込み

「手順1:PySparkでデータをDataframeとして読み込み」に関してご紹介します。

PySparkでは、以下の「spark.read.json」でJsonファイルをDataframeとして読み込めます。

from pyspark.sql import DataFrameReader
from pyspark.sql import SparkSession
from pyspark.sql.functions import max as max_, col, when
import sys
import json

# 1. 引数を変数に格納
args = sys.argv

## 2. sparkの設定を作成
spark = SparkSession \
    .builder \
    .appName("concat json") \
    .getOrCreate()
spark_context = spark.sparkContext

## 3. jsonファイルからDataframeを作成
users_df   = spark.read.json("rocketchat_data/users.json")
message_df = spark.read.json("rocketchat_data/message.json")


spark.read.jsonの引数として、HDFS上のJsonファイルPathを指定します。


以上で、「手順1:PySparkでデータをDataframeとして読み込み」は完了です。

手順2:PySparkのDataframeを結合(join)

「手順2:PySparkのDataframeを結合(join)」に関してご紹介します。

PySparkでは、以下の「join」メソッドを利用することで2つのDataframeを結合(join)できます。

## 4. 外部結合を実施
joined_df = message_df.join(users_df, message_df.u._id == users_df._id, "inner")
joined_df.show()


joinメソッドの引数は以下となります。
・第1引数:結合(join)したいDataframeを指定
・第2引数:結合(join)条件を指定
・第3引数:結合(join)の種類を指定(18種類)
参照先:Sparkのdataframeに関する公式サイト


以上で、「手順2:PySparkのDataframeを結合(join)」は完了です。

手順3:PySparkでDataframeを結合(join)するプログラムを実行

「手順3:PySparkでDataframeを結合(join)するプログラムを実行」に関してご紹介します。

先ほどのプログラムをtest.pyとし、「spark-submit」コマンドを利用し実行します。

Sparkが導入されている環境で、以下のコマンドを実行しましょう。

$ SPARK_CMD="/opt/oss/spark-2.4.5-bin-without-hadoop/bin/spark-submit"
$ ${SPARK_CMD} --master yarn --conf "spark.executor.memoryOverhead=600" \
             --num-executors 2 \
             --executor-cores 1 \
             --executor-memory 3G \
             test.py


「spark-submit」コマンドの出力結果として、以下のように結合(join)したDataframeのスキーマ情報が表示されます。

★結合したDataframeのスキーマ情報(以降、スキーマ1と表記)★

root
 |-- _id: string (nullable = true)
 |-- _updatedAt: struct (nullable = true)
 |    |-- $date: string (nullable = true)
 |-- attachments: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- image_dimensions: struct (nullable = true)
 |    |    |    |-- height: long (nullable = true)
 |    |    |    |-- width: long (nullable = true)
 |    |    |-- image_preview: string (nullable = true)
 |    |    |-- image_size: long (nullable = true)
 |    |    |-- image_type: string (nullable = true)
 |    |    |-- image_url: string (nullable = true)
 |    |    |-- title: string (nullable = true)
 |    |    |-- title_link: string (nullable = true)
 |    |    |-- title_link_download: boolean (nullable = true)
 |    |    |-- type: string (nullable = true)
 |-- channels: array (nullable = true)
 |    |-- element: string (containsNull = true)
 |-- file: struct (nullable = true)
 |    |-- _id: string (nullable = true)
 |    |-- name: string (nullable = true)
 |    |-- type: string (nullable = true)
 |-- groupable: boolean (nullable = true)
 |-- mentions: array (nullable = true)
 |    |-- element: string (containsNull = true)
 |-- msg: string (nullable = true)
 |-- reactions: struct (nullable = true)
 |    |-- :grinning:: struct (nullable = true)
 |    |    |-- usernames: array (nullable = true)
 |    |    |    |-- element: string (containsNull = true)
 |-- replies: array (nullable = true)
 |    |-- element: string (containsNull = true)
 |-- rid: string (nullable = true)
 |-- t: string (nullable = true)
 |-- tcount: long (nullable = true)
 |-- tlm: struct (nullable = true)
 |    |-- $date: string (nullable = true)
 |-- tmid: string (nullable = true)
 |-- ts: struct (nullable = true)
 |    |-- $date: string (nullable = true)
 |-- tshow: boolean (nullable = true)
 |-- u: struct (nullable = true)
 |    |-- _id: string (nullable = true)
 |    |-- username: string (nullable = true)
 |-- __rooms: array (nullable = true)
 |    |-- element: string (containsNull = true)
 |-- _id: string (nullable = true)
 |-- _updatedAt: struct (nullable = true)
 |    |-- $date: string (nullable = true)
 |-- active: boolean (nullable = true)
 |-- avatarETag: string (nullable = true)
 |-- avatarOrigin: string (nullable = true)
 |-- createdAt: struct (nullable = true)
 |    |-- $date: string (nullable = true)
 |-- emails: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- address: string (nullable = true)
 |    |    |-- verified: boolean (nullable = true)
 |-- lastLogin: struct (nullable = true)
 |    |-- $date: string (nullable = true)
 |-- name: string (nullable = true)
 |-- roles: array (nullable = true)
 |    |-- element: string (containsNull = true)
 |-- services: struct (nullable = true)
 |    |-- email: struct (nullable = true)
 |    |    |-- verificationTokens: array (nullable = true)
 |    |    |    |-- element: struct (containsNull = true)
 |    |    |    |    |-- address: string (nullable = true)
 |    |    |    |    |-- token: string (nullable = true)
 |    |    |    |    |-- when: struct (nullable = true)
 |    |    |    |    |    |-- $date: string (nullable = true)
 |    |-- email2fa: struct (nullable = true)
 |    |    |-- changedAt: struct (nullable = true)
 |    |    |    |-- $date: string (nullable = true)
 |    |    |-- enabled: boolean (nullable = true)
 |    |-- password: struct (nullable = true)
 |    |    |-- bcrypt: string (nullable = true)
 |    |    |-- reset: struct (nullable = true)
 |    |    |    |-- email: string (nullable = true)
 |    |    |    |-- reason: string (nullable = true)
 |    |    |    |-- token: string (nullable = true)
 |    |    |    |-- when: struct (nullable = true)
 |    |    |    |    |-- $date: string (nullable = true)
 |    |-- resume: struct (nullable = true)
 |    |    |-- loginTokens: array (nullable = true)
 |    |    |    |-- element: struct (containsNull = true)
 |    |    |    |    |-- hashedToken: string (nullable = true)
 |    |    |    |    |-- when: struct (nullable = true)
 |    |    |    |    |    |-- $date: string (nullable = true)
 |-- status: string (nullable = true)
 |-- statusConnection: string (nullable = true)
 |-- statusDefault: string (nullable = true)
 |-- type: string (nullable = true)
 |-- username: string (nullable = true)
 |-- utcOffset: long (nullable = true)


以上で、「手順3:PySparkでDataframeを結合(join)するプログラムを実行」は完了です。

上記の3手順で、PySparkのDataframeで結合(join)できました。


結合(join)したPySparkのDataframeから特定列を抽出する手順

「結合(join)したPySparkのDataframeから特定列を抽出する手順」に関してご紹介します。

★結合(join)したPySparkのDataframeから特定列を抽出する流れ★
手順1:PySparkでネストされたJsonをDataframeとして読み込み
手順2:結合(join)したPySparkのDataframeから特定列を抽出
手順3:結合(join)したPySparkのDataframeから特定列を抽出するプログラムを実行


上記の3手順で、結合(join)したPySparkのDataframeから特定列を抽出できます。

以降で、「結合(join)したPySparkのDataframeから特定列を抽出する流れ」の各手順に関して説明します。

手順1:PySparkでネストされたJsonをDataframeとして読み込み

「手順1:PySparkでネストされたJsonをDataframeとして読み込み」に関してご紹介します。

「spark.read.json」を使うことで、PySparkではネストされたJsonもDataframeとして読み込めます。

今回は、複雑なネスト構造をしている「PySparkのDataframeを3手順で結合(join)する手順」で結合(join)したDataframeを利用します。

以上で、「手順1:PySparkでネストされたJsonをDataframeとして読み込み」は完了です。

手順2:結合(join)したPySparkのDataframeから特定列を抽出

「手順2:結合(join)したPySparkのDataframeから特定列を抽出」に関してご紹介します。

PySparkでDataframeから特定列を抽出したい場合、「select/col/getItem」を利用します。

・selectメソッド
 → Dataframeから指定したカラムを抽出します。
・colメソッド
 → 指定したカラムをTypedColumn型のオブジェクトとして扱います。
・getItemメソッド
 → カラムに配列データが格納されている場合、
   getItemメソッドを使うことで指定した要素を抽出できます。


スキーマ1の下記情報を抽出してみましょう。
・username
・msg
・emailsの1番目のaddress

上記情報を抽出するコードは、以下となります。

## 5. ネストされた情報を抽出し表示
joined_df.select("username","msg", col("emails").getItem(0)["address"]).show()


以上で、「手順2:結合(join)したPySparkのDataframeから特定列を抽出」は完了です。

手順3:結合(join)したPySparkのDataframeから特定列を抽出するプログラムを実行

「手順3:結合(join)したPySparkのDataframeから特定列を抽出するプログラムを実行」に関してご紹介します。

test.pyに手順2で紹介した「ネストされた情報を抽出し表示」のコードを追記し、「spark-submit」コマンドを実行します。

Sparkが導入されている環境で以下のコマンドを実行しましょう。

$ SPARK_CMD="/opt/oss/spark-2.4.5-bin-without-hadoop/bin/spark-submit"
$ ${SPARK_CMD} --master yarn --conf "spark.executor.memoryOverhead=600" \
             --num-executors 2 \
             --executor-cores 1 \
             --executor-memory 3G \
             test.py


「spark-submit」コマンドの実行結果として、指定した情報のみ出力できます。

+--------+----------------+--------------------+
|username|             msg|   emails[0].address|
+--------+----------------+--------------------+
| xxxxxxx|         yyyyyyy|xxxxxxxxxxxxxxxx@...|
| xxxxxxx|                |xxxxxxxxxxxxxxxx@...|
| xxxxxxx|わかりましたか?|xxxxxxxxxxxxxxxx@...|
| xxxxxxx|               2|xxxxxxxxxxxxxxxx@...|
| xxxxxxx|               1|xxxxxxxxxxxxxxxx@...|
| xxxxxxx|               3|xxxxxxxxxxxxxxxx@...|
| xxxxxxx|            YES!|xxxxxxxxxxxxxxxx@...|
| xxxxxxx|            test|xxxxxxxxxxxxxxxx@...|
| xxxxxxx|          da!!!!|xxxxxxxxxxxxxxxx@...|
+--------+----------------+--------------------+


以上で、「手順3:結合(join)したPySparkのDataframeから特定列を抽出するプログラムを実行」は完了です。

上記の3手順で、「結合(join)したPySparkのDataframeから特定列を抽出する手順」ができました。


【まとめ】PySparkのDataframeを3手順で結合(join)する

今回の記事を通して、PySparkのDataframeを3手順で結合(join)するをご紹介することで、以下の悩みを解消しました。

★悩み★
・PySparkのDataframeで結合(join)はどうすればいいのだろうか。
・結合(join)したPySparkのDataframeから特定列の抽出手順を知りたい。


「PySparkでDataframeを結合(join)する方法は?」と悩んでいるあなたにこの記事が少しでも役に立てれば幸いです。

コメント

タイトルとURLをコピーしました