More than 5 years have passed since last update.

Spark DataframeのSample Code集

Last updated at 2019-04-18Posted at 2016-02-16

はじめに：Spark Dataframeとは

Spark Ver 1.3からSpark Dataframeという機能が追加されました。特徴として以下の様な物があります。

Spark RDDにSchema設定を加えると、Spark DataframeのObjectを作成できる
Dataframeの利点は、
- SQL風の文法で、条件に該当する行を抽出したり、Dataframe同士のJoinができる
- filter, selectというmethodで、条件に該当する行、列を抽出できる
- groupBy → aggというmethodで、Logの様々な集計ができる
- UDF(User Defined Function)で独自関数で列に処理ができる
- SQLで言うPivotもサポート (Spark v1.6からの機能)

つまり、RDDのmapやfilterでシコシコ記述するよりもSimple Codeで、且つ高速に処理が行えるのがウリです。Dataの前処理はRDDでやるとして、さっさとDataframeに読み込んだ方がmajiで捗ります。Dataframeのメモが散在したので、備忘録がてらSample codeをチラ裏しておきます。

なお、

この記事にあるAPI以外にも、沢山APIがあります
この記事のIPython Notebookをここに置いておくので、随時実行してみて下さい
Spark Dataframeの縮約/集計のSample Code

Sample Logの読み込み

Access Logを題材として使います。技術評論社さんの本で使われていたAccess Log(csv)で、csv fileへの直リンはこちらです。csvの中身は、日付、User_ID, Campaign_IDの3つの情報を持つ以下の様なLogです

click.at	user.id	campaign.id
2015/4/27 20:40	144012	Campaign077
2015/4/27 0:27	24485	Campaign063
2015/4/27 0:28	24485	Campaign063
2015/4/27 0:33	24485	Campaign038

csvを読み込んでRDDにします。1行目のheaderの削除と、1列目をdatetime Objectとして読み込みます。

import json, os, datetime, collections, commands
from pyspark.sql import SQLContext, Row
from pyspark.sql.types import *

if not os.path.exists("./click_data_sample.csv"):
    print "csv file not found at master node, will download and copy to HDFS"
    commands.getoutput("wget -q http://image.gihyo.co.jp/assets/files/book/2015/978-4-7741-7631-4/download/click_data_sample.csv")
    commands.getoutput("hadoop fs -copyFromLocal -f ./click_data_sample.csv /user/hadoop/")

whole_raw_log = sc.textFile("/user/hadoop/click_data_sample.csv")
header = whole_raw_log.first()
whole_log = whole_raw_log.filter(lambda x:x !=header).map(lambda line: line.split(","))\
            .map(lambda line: [datetime.datetime.strptime(line[0].replace('"', ''), '%Y-%m-%d %H:%M:%S'), int(line[1]), line[2].replace('"', '')])

whole_log.take(3)
#[[datetime.datetime(2015, 4, 27, 20, 40, 40), 144012, u'Campaign077'],
# [datetime.datetime(2015, 4, 27, 0, 27, 55), 24485, u'Campaign063'],
# [datetime.datetime(2015, 4, 27, 0, 28, 13), 24485, u'Campaign063']]

Dataframeの作成方法

RDDから作成

Dataframeは、元となるRDDがあれば、Columnの名前とそれぞれのType(TimestampType, IntegerType, StringTypeなど)を指定して、sqlContext.createDataFrame(my_rdd, my_schema)で作成できます。Schemaの定義はここを参照。

printSchema(), dtypesでSchema情報、count()で行数、show(n)で最初のn件のrecordの表示です。

fields = [StructField("access_time", TimestampType(), True), StructField("userID", IntegerType(), True), StructField("campaignID", StringType(), True)]
schema = StructType(fields)

whole_log_df = sqlContext.createDataFrame(whole_log, schema)
print whole_log_df.count()
print whole_log_df.printSchema()
print whole_log_df.dtypes
print whole_log_df.show(5)

#327430
#root
# |-- access_time: timestamp (nullable = true)
# |-- userID: integer (nullable = true)
# |-- campaignID: string (nullable = true)
#
#[('access_time', 'timestamp'), ('userID', 'int'), ('campaignID', 'string')]
#
#+--------------------+------+-----------+
#|         access_time|userID| campaignID|
#+--------------------+------+-----------+
#|2015-04-27 20:40:...|144012|Campaign077|
#|2015-04-27 00:27:...| 24485|Campaign063|
#|2015-04-27 00:28:...| 24485|Campaign063|
#|2015-04-27 00:33:...| 24485|Campaign038|
#|2015-04-27 01:00:...| 24485|Campaign063|

csv fileから直接作成

csvから読み込んだdataをそのままDataframeにするには、Spark Packageの1つであるspark-csvを使うと楽です。特に指定しないと全てstringとして読み込みますが、inferSchemaを指定してあげると良い感じに類推してくれます。

whole_log_df_2 = sqlContext.read.format("com.databricks.spark.csv").option("header", "true").load("/user/hadoop/click_data_sample.csv")
print whole_log_df_2.printSchema()
print whole_log_df_2.show(5)

#root
# |-- click.at: string (nullable = true)
# |-- user.id: string (nullable = true)
# |-- campaign.id: string (nullable = true)
#
#+-------------------+-------+-----------+
#|           click.at|user.id|campaign.id|
#+-------------------+-------+-----------+
#|2015-04-27 20:40:40| 144012|Campaign077|
#|2015-04-27 00:27:55|  24485|Campaign063|
#|2015-04-27 00:28:13|  24485|Campaign063|
#|2015-04-27 00:33:42|  24485|Campaign038|
#|2015-04-27 01:00:04|  24485|Campaign063|

whole_log_df_3 = sqlContext.read.format("com.databricks.spark.csv").option("header", "true").option("inferSchema", "true").load("/user/hadoop/click_data_sample.csv")
print whole_log_df_3.printSchema()

#root
# |-- click.at: timestamp (nullable = true)
# |-- user.id: integer (nullable = true)
# |-- campaign.id: string (nullable = true)

ちなみに、column名に.が入って居ると色々面倒なので、withColumnRenamedでrename可能です(renameした別のDataframeを作成可能です)。

whole_log_df_4 = whole_log_df_3.withColumnRenamed("click.at", "access_time")\
                 .withColumnRenamed("user.id", "userID")\
                 .withColumnRenamed("campaign.id", "campaignID")
print whole_log_df_4.printSchema()

#root
# |-- access_time: timestamp (nullable = true)
# |-- userID: integer (nullable = true)
# |-- campaignID: string (nullable = true)

jsonから直接作成

json fileから読み込んだdataをそのままDataframeにするにはsqlContext.read.jsonを使います。fileの各行を1 json objectとして扱います、存在しないKeyがある場合には、nullが入ります。

# test_json.json contains following 3 lines, last line doesn't have "campaignID" key
#
#{"access_time": "2015-04-27 20:40:40", "userID": "24485", "campaignID": "Campaign063"}
#{"access_time": "2015-04-27 00:27:55", "userID": "24485", "campaignID": "Campaign038"}
#{"access_time": "2015-04-27 00:27:55", "userID": "24485"}

df_json = sqlContext.read.json("/user/hadoop/test_json.json")
df_json.printSchema()
df_json.show(5)

#root
# |-- access_time: string (nullable = true)
# |-- campaignID: string (nullable = true)
# |-- userID: string (nullable = true)
#
#+-------------------+-----------+------+
#|        access_time| campaignID|userID|
#+-------------------+-----------+------+
#|2015-04-27 20:40:40|Campaign063| 24485|
#|2015-04-27 00:27:55|Campaign038| 24485|
#|2015-04-27 00:27:55|       null| 24485|
#+-------------------+-----------+------+

parquetから直接作成

parquet fileから読み込んだdataをそのままDataframeにするにはsqlContext.read.parquetを使います。parquet fileが置いてあるFolderを指定すると、そのFolder以下のparquet fileを一括で読み込んでくれます。

sqlContext.read.parquet("/user/hadoop/parquet_folder/")

SQL文でQuery

Dataframeに対して、SQLの文でQueryを掛けるSampleです。registerTempTableでDataframeにSQL Table nameを付与すると、SQLのTable名として参照できます。sqlContext.sql(SQL文)の戻り値もDataframeです。

なお、Sub Queryを記載する事も可能なのですが、Sub Query側にAliasを付与しないと、何故かSyntax errorが起きるので注意です。

#単純なSQL query

whole_log_df.registerTempTable("whole_log_table")

print sqlContext.sql(" SELECT * FROM whole_log_table where campaignID == 'Campaign047' ").count()
#18081
print sqlContext.sql(" SELECT * FROM whole_log_table where campaignID == 'Campaign047' ").show(5)
#+--------------------+------+-----------+
#|         access_time|userID| campaignID|
#+--------------------+------+-----------+
#|2015-04-27 05:26:...| 14151|Campaign047|
#|2015-04-27 05:26:...| 14151|Campaign047|
#|2015-04-27 05:26:...| 14151|Campaign047|
#|2015-04-27 05:27:...| 14151|Campaign047|
#|2015-04-27 05:28:...| 14151|Campaign047|
#+--------------------+------+-----------+


#SQL文の中に変数を入れる場合
for count in range(1, 3):
    print "Campaign00" + str(count)
    print sqlContext.sql("SELECT count(*) as access_num FROM whole_log_table where campaignID == 'Campaign00" + str(count) + "'").show()

#Campaign001
#+----------+
#|access_num|
#+----------+
#|      2407|
#+----------+
#
#Campaign002
#+----------+
#|access_num|
#+----------+
#|      1674|
#+----------+

#Sub Queryの場合：
print sqlContext.sql("SELECT count(*) as first_count FROM (SELECT userID, min(access_time) as first_access_date FROM whole_log_table GROUP BY userID) subquery_alias WHERE first_access_date < '2015-04-28'").show(5)
#+------------+
#|first_count |
#+------------+
#|       20480|
#+------------+

filter, selectで条件付き検索

Dataframeに対しての簡易的な検索機能です。上記にあるSQL文でQueryと機能は似ていますが、filter, selectは簡易的な検索機能という位置づけです。filterは条件に該当する行の抽出、selectは列を抽出します。RDDのfilterとちょっと文法が違うのに注意です。

#Sample for filter
print whole_log_df.filter(whole_log_df["access_time"] < "2015-04-28").count()
#41434
print whole_log_df.filter(whole_log_df["access_time"] > "2015-05-01").show(3)
#+--------------------+------+-----------+
#|         access_time|userID| campaignID|
#+--------------------+------+-----------+
#|2015-05-01 22:11:...|114157|Campaign002|
#|2015-05-01 23:36:...| 93708|Campaign055|
#|2015-05-01 22:51:...| 57798|Campaign046|
#+--------------------+------+-----------+

#Sample for select
print whole_log_df.select("access_time", "userID").show(3)
#+--------------------+------+
#|         access_time|userID|
#+--------------------+------+
#|2015-04-27 20:40:...|144012|
#|2015-04-27 00:27:...| 24485|
#|2015-04-27 00:28:...| 24485|
#+--------------------+------+

groupByで集計

groupByは、RDDのreduceByKeyに似た機能を提供しますが、groupByはここにあるmethodをその後ろでCallする事で、様々な集計機能を実現できます。代表的なのはaggとcountです。

groupBy→countで集計

campaignIDをKeyにしてgroupByを実行し、そのRecord数をcount()で集計してくれます。groupByに複数のKeyを列挙すれば、その組み合わせをkeyとしてgroupByしてくれます。

print whole_log_df.groupBy("campaignID").count().sort("count", ascending=False).show(5)
#+-----------+-----+
#| campaignID|count|
#+-----------+-----+
#|Campaign116|22193|
#|Campaign027|19206|
#|Campaign047|18081|
#|Campaign107|13295|
#|Campaign131| 9068|
#+-----------+-----+

print whole_log_df.groupBy("campaignID", "userID").count().sort("count", ascending=False).show(5)
#+-----------+------+-----+
#| campaignID|userID|count|
#+-----------+------+-----+
#|Campaign047| 30292|  633|
#|Campaign086|107624|  623|
#|Campaign047|121150|  517|
#|Campaign086| 22975|  491|
#|Campaign122| 90714|  431|
#+-----------+------+-----+

groupBy→aggで集計

userIDをKeyにしてGroupByを実行し、その集計結果の平均や最大/最小を計算が可能です。agg({key:value})で、keyの列に対してvalueの関数(min,sum, ave etc)を実行した結果を返します。戻り値はDataframeなので、.filter()で更に行を絞る事も可能です。

print whole_log_df.groupBy("userID").agg({"access_time": "min"}).show(3)
#+------+--------------------+
#|userID|    min(access_time)|
#+------+--------------------+
#|  4831|2015-04-27 22:49:...|
#| 48631|2015-04-27 22:15:...|
#|143031|2015-04-27 21:52:...|
#+------+--------------------+

print whole_log_df.groupBy("userID").agg({"access_time": "min"}).filter("min(access_time) < '2015-04-28'").count()
#20480

groupBy→pivotで縦横変換

PivotはSpark v1.6からの新機能でSQLのPivotと似た機能を提供します。Sample codeのPivotの場合、以下の様に縦横が変化します。

pivot前(agged_df)
- 行数が(UserID数 (=75,545) x campainID数 (=133) )
- 列が3列
pivot後(pivot_df)
- 行数がUserID数 (=75,545)
- 列が UserID + CampainID数 = 1 + 133 = 134

必ず、groupBy("縦のままの列").pivot("縦から横へ変換したい列").sum("集計値の列")と3つのmethodをchainで呼ぶ必要があります。

agged_df = whole_log_df.groupBy("userID", "campaignID").count()
print agged_df.show(3)

#+------+-----------+-----+
#|userID| campaignID|count|
#+------+-----------+-----+
#|155812|Campaign107|    4|
#|103339|Campaign027|    1|
#|169114|Campaign112|    1|
#+------+-----------+-----+

#値が無いCellは、nullが入る
pivot_df = agged_df.groupBy("userID").pivot("campaignID").sum("count")
print pivot_df.printSchema()

#root
# |-- userID: integer (nullable = true)
# |-- Campaign001: long (nullable = true)
# |-- Campaign002: long (nullable = true)
# ..
# |-- Campaign133: long (nullable = true)

#値が無いCellを0で埋めたい場合
pivot_df2 = agged_df.groupBy("userID").pivot("campaignID").sum("count").fillna(0)

UDFで列の追加

Spark DataframeではUDFが使えます、主な用途は、列の追加になるかと思います。Dataframeは基本Immutable(不変)なので、列の中身の変更はできず、列を追加した別のDataframeを作成する事になります。

from pyspark.sql.functions import UserDefinedFunction
from pyspark.sql.types import DoubleType

def add_day_column(access_time):
    return int(access_time.strftime("%Y%m%d"))
    
my_udf = UserDefinedFunction(add_day_column, IntegerType())
print whole_log_df.withColumn("access_day", my_udf("access_time")).show(5)

#+--------------------+------+-----------+----------+
#|         access_time|userID| campaignID|access_day|
#+--------------------+------+-----------+----------+
#|2015-04-27 20:40:...|144012|Campaign077|  20150427|
#|2015-04-27 00:27:...| 24485|Campaign063|  20150427|
#|2015-04-27 00:28:...| 24485|Campaign063|  20150427|
#|2015-04-27 00:33:...| 24485|Campaign038|  20150427|
#|2015-04-27 01:00:...| 24485|Campaign063|  20150427|
#+--------------------+------+-----------+----------+

UDFの表記は、lambda関数を使って書くことも可能です。

my_udf2 = UserDefinedFunction(lambda x: x + 5, IntegerType())
print whole_log_df.withColumn("userID_2", my_udf2("userID")).show(5)

#+--------------------+------+-----------+--------+
#|         access_time|userID| campaignID|userID_2|
#+--------------------+------+-----------+--------+
#|2015-04-27 20:40:...|144012|Campaign077|  144017|
#|2015-04-27 00:27:...| 24485|Campaign063|   24490|
#|2015-04-27 00:28:...| 24485|Campaign063|   24490|
#|2015-04-27 00:33:...| 24485|Campaign038|   24490|
#|2015-04-27 01:00:...| 24485|Campaign063|   24490|
#+--------------------+------+-----------+--------+

逆に、特定の列を削除したいDataframeを作るにはdf.drop()を使います。

print whole_log_df.drop("userID").show(3)

#+--------------------+-----------+
#|         access_time| campaignID|
#+--------------------+-----------+
#|2015-04-27 20:40:...|Campaign077|
#|2015-04-27 00:27:...|Campaign063|
#|2015-04-27 00:28:...|Campaign063|
#+--------------------+-----------+

Joinで2つのDataframeを結合させる

2つのDataframeをJoinさせる事も可能です。ここでは、Heavy User(Access数が100回以上あるUser)のLogのみを全体のLogから抽出するケースを考えてみます。

まず、Access数が100回以上あるUserのUser IDとそのAccess数を、.groupBy("userID").count()で集計し、filterで100回以上に絞り込みます。

heavy_user_df1 = whole_log_df.groupBy("userID").count()
heavy_user_df2 = heavy_user_df1.filter(heavy_user_df1 ["count"] >= 100)

print heavy_user_df2 .printSchema()
print heavy_user_df2 .show(3)
print heavy_user_df2 .count()

#root
# |-- userID: integer (nullable = true)
# |-- count: long (nullable = false)
#
#+------+-----+
#|userID|count|
#+------+-----+
#| 84231|  134|
#| 13431|  128|
#|144432|  113|
#+------+-----+
#
#177

元のDataframe(こちらがLeftになる)でjoin methodを呼び、joinの相手(Rightになる)とjoinの条件を書くと、SQLのjoinの様にDataframeの結合が可能です。

joinの形式は、inner, outer, left_outer, rignt_outerなどが選べるはずなのですが、inner以外は意図した挙動で動いてくれない為(しかもouterとして処理される)、とりあえずinnerで外部結合して後にdropで要らないColumnを削除するようにしています。詳細Option等は公式Pageを参照して下さい。

以下のjoin処理により、Access数が100回以上あるUser(177名)に該当する38,729行のLogを取り出す事ができました(全体のLogは約32万行)。

joinded_df = whole_log_df.join(heavy_user_df2, whole_log_df["userID"] == heavy_user_df2["userID"], "inner").drop(heavy_user_df2["userID"]).drop("count")
print joinded_df.printSchema()
print joinded_df.show(3)
print joinded_df.count()

#root
# |-- access_time: timestamp (nullable = true)
# |-- campaignID: string (nullable = true)
# |-- userID: integer (nullable = true)

#None
#+--------------------+-----------+------+
#|         access_time| campaignID|userID|
#+--------------------+-----------+------+
#|2015-04-27 02:07:...|Campaign086| 13431|
#|2015-04-28 00:07:...|Campaign086| 13431|
#|2015-04-29 06:01:...|Campaign047| 13431|
#+--------------------+-----------+------+
#
#38729

Dataframeから列を取り出す

列のLabelを取り出すには、df.columnsで列のLabelのList(not Dataframe)が取り出せる
特定の列を取り出すには、df.select("userID").map(lambda x: x[0]).collect()で、"userID"列のList(not RDD/Dataframe)が取り出せる
特定の列で重複が無いListを取り出すには.distinct()をDataframeの最後に追加すればOK

print whole_log_df.columns
#['access_time', 'userID', 'campaignID']

print whole_log_df.select("userID").map(lambda x: x[0]).collect()[:5]
#[144012, 24485, 24485, 24485, 24485]

print whole_log_df.select("userID").distinct().map(lambda x:x[0]).collect()[:5]
#[4831, 48631, 143031, 39631, 80831]

DataframeからRDD/Listに戻す

DataframeをRDDに戻すには、大きく2つの方法があります

.mapを呼ぶ
- DataframeのSchema情報は破棄され、Dataframeの各行がそれぞれlistになったRDDに変換されます
.rddを呼ぶ
- Dataframeの各行がそれぞれRow OjbectなRDDに変換されます。Row ObjectはSpark SQLで一行分のデータを保持する為のObjectです
- my_rdd.rdd.map(lambda x:x.asDict())と、Row objectに対して.asDict()を呼んであげると、Key-ValueなRDDに変換可能です

#convert to rdd by ".map"
print whole_log_df.groupBy("campaignID").count().map(lambda x: [x[0], x[1]]).take(5)
#[[u'Campaign033', 786], [u'Campaign034', 3867], [u'Campaign035', 963], [u'Campaign036', 1267], [u'Campaign037', 1010]]

# rdd -> normal list can be done with "collect".
print whole_log_df.groupBy("campaignID").count().map(lambda x: [x[0], x[1]]).collect()[:5]
#[[u'Campaign033', 786], [u'Campaign034', 3867], [u'Campaign035', 963], [u'Campaign036', 1267], [u'Campaign037', 1010]]

#convert to rdd by ".rdd" will return "Row" object
print whole_log_df.groupBy("campaignID").rdd.take(3)
#[Row(campaignID=u'Campaign033', count=786), Row(campaignID=u'Campaign034', count=3867), Row(campaignID=u'Campaign035', count=963)]

#`.asDict()` will convert to Key-Value RDD from Row object
print whole_log_df.groupBy("campaignID").rdd.map(lambda x:x.asDict()).take(3)
#[{'count': 786, 'campaignID': u'Campaign033'}, {'count': 3867, 'campaignID': u'Campaign034'}, {'count': 963, 'campaignID': u'Campaign035'}]

DataframeからParquet fileに書き出す

DataframeをParquet形式でfileに書き出せば、schema情報を保持したままfileにExportが可能です。なお、ExportするS3 bucketのdirectoryが既に存在する場合には書き込みが失敗します、まだ存在していないDirectory名を指定して下さい。

#write to parquet filed
whole_log_df.select("access_time", "userID").write.parquet("s3n://my_S3_bucket/parquet_export") 

#reload from parquet filed
reload_df = sqlContext.read.parquet("s3n://my_S3_bucket/parquet_export") 
print reload_df.printSchema()

165

173

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up