Help us understand the problem. What is going on with this article?

Spark API チートシート

目的

Sparkのよく使うAPIを(主に自分用に)メモしておくことで、久しぶりに開発するときでもサクサク使えるようにしたい。とりあえずPython版をまとめておきます(Scala版も時間があれば加筆するかも)

このチートシートはあくまでチートシートなので(引数が省略してあったりします)、時間がある方はきちんと公式APIドキュメント(Spark Python API Docs)を見て下さい。

Spark API チートシート(Python)

以下では次を前提とする

from pyspark import SparkContext
from pyspark.sql import SQLContext

sc = SparkContext()
sqlContext = SQLContext(sc)

RDD

RDDを作る(データの読み込み)

parallelize

sc.parallelize(collection) リストやタプルからRDDを作る

>>> a = [1, 2, 3, 4, 5]
>>> rdd = sc.parallelize(a)
textFile

sc.textFile(file) ファイルを読み込む。ワイルドカードや正規表現も使える。

>>> rdd = sc.textFile("./words.txt")
wholeTextFiles

sc.wholeTextFiles(dierctory) ディレクトリの各ファイルの内容全体をそれぞれRDDの一つの要素に入力する

# $ ls
# a.json b.json c.json
>>> rdd = sc.textWholeFiles("./")

Action

Actionが実行されると初めてTransformationが順に実行される(遅延実行)

要素を返すもの

collect

collect() 全ての要素を返す

>>> print(rdd.collect())
[1, 2, 3, 4, 5]
take

take(n) 最初n個の要素を返す

>>> print(rdd.take(3))
[1, 2, 3]
first

first() 一番最初の要素を返す

>>> rdd = sc.parallelize([1, 2, 3])
>>> rdd.first()
1
top

top(n) 大きいものからn個要素を返す

>>> rdd = sc.parallelize([1, 2, 3])
>>> rdd.top(2)
[3, 2]

(統計)量を返すもの

count

count() 要素数を数えて返す

>>> rdd = sc.parallelize([1, 2, 3])
>>> rdd.count()
3
mean

mean() 平均を返す

>>> rdd = sc.parallelize([1, 2, 3])
>>> rdd.mean()
3.0
sum

sum() 合計を返す

>>> rdd = sc.parallelize([1, 2, 3])
>>> rdd.sum()
6
variance

variance() 分散を返す

>>> rdd = sc.parallelize([1, 2, 3])
>>> rdd.variance()
0.6666666666666666
stdev

stdev() 標準偏差を返す

>>> rdd = sc.parallelize([1, 2, 3])
>>> rdd.stdev()
0.816496580927726

保存するもの

saveAsTextFile

saveAsTextFile(file) ファイルを保存する

>>> rdd.saveAsTextFile("./a.txt")

Transformation

Transformationはimmutableな新しいRDDを返す

filter/map/reduce

filter

filter(f) fが真となる要素だけを含むRDDを返す

>>> rdd = sc.parallelize([1, 2, 3])
>>> rdd.filter(lambda x: x % 2 == 0).collect()
[2]
map

map(f) 全ての要素にfを作用させたRDDを返す

>>> rdd = sc.parallelize([1, 2, 3])
>>> rdd.map(lambda x: x * 2).collect()
[2, 4, 6]
flatMap

flatMap(f) 全ての要素にfを作用させたあと、要素内のリストを展開したRDDを返す

>>> rdd = sc.parallelize(["This is a pen", "This is an apple"])
>>> rdd.flatMap(lambda x: x.split()).collect()
['This', 'is', 'a', 'pen', 'This', 'is', 'an', 'apple']
Reduce

reduce(f) 二つの要素にfを作用させ続けて一つの返り値を得る

>>> rdd = sc.parallelize([1, 2, 3])
>>> rdd.reduce(lambda x, y: x + y)
6

ペアRDDに対する操作

ペアRDDをつくる

ペアRDDはPythonでいうTupleを要素に持つRDD。keyとvalueを扱うことができる。
作り方はkeyByを使うかmapで要素数2のタプルを要素に返す。

keyBy(PairRDD)

keyBy(f) 普通のRDDの要素にfを作用させ、その返り値をkeyに、元の要素をそのままvalueにしたRDDを返す

>>> rdd = sc.parallelize(["Ken 27 180 83", "Bob 32 170 65", "Meg 29 165 45"])
>>> rdd.keyBy(lambda x: x.split()[0]).collect()
[('Ken', 'Ken 27 180 83'), ('Bob', 'Bob 32 170 65'), ('Meg', 'Meg 29 165 45')]
keys

keys ペアRDDのkeyだけからなるRDDを返す

>>> rdd = sc.parallelize([("Ken", 2), ("Bob", 3), ("Taka", 1), ("Ken", 3), ("Bob", 2)])
>>> rdd.keys().collect()
['Ken', 'Bob', 'Taka', 'Ken', 'Bob']
values

values ペアRDDのvlaueだけからなるRDDを返す

>>> rdd = sc.parallelize([("Ken", 2), ("Bob", 3), ("Taka", 1), ("Ken", 3), ("Bob", 2)])
>>> rdd.values().collect()
[2, 3, 1, 3, 2]
flatMapValues

flatMapValues(f) PairRDDのvalueにflatMapを作用させてkeyを複製して所謂縦持ちにする

>>> rdd = sc.parallelize([("Ken", "Yumi,Yukiko"), ("Bob", "Meg, Tomomi, Akira"), ("Taka", "Yuki")])
>>> rdd.flatMapValues(lambda x: x.split(","))
[('Ken', 'Yumi'),
 ('Ken', 'Yukiko'),
 ('Bob', 'Meg'),
 ('Bob', ' Tomomi'),
 ('Bob', ' Akira'),
 ('Taka', 'Yuki')]
reduceByKey

reduceByKey(f) 同じkeyの要素でグループ化してvalueにreduceを作用させる

>>> rdd = sc.parallelize([("Ken", 2), ("Bob", 3), ("Taka", 1), ("Ken", 3), ("Bob", 2)])
>>> rdd.reduceByKey(lambda x, y: x + y).collect()
[('Taka', 1), ('Bob', 5), ('Ken', 5)]
countByKey

countByKey() 同じkeyの値がいくつあるか数えてdictで返す

>>> rdd = sc.parallelize([("Ken", 2), ("Bob", 3), ("Taka", 1), ("Ken", 3), ("Bob", 2)])
>>> rdd.countByKey()
defaultdict(<type 'int'>, {'Ken': 2, 'Bob': 2, 'Taka': 1})
sortByKey

sortByKey ペアRDDをkeyでソートします

>>> rdd = sc.parallelize([("cba", 2), ("abc", 3), ("bac", 1), ("bbb", 
>>> rdd.sortByKey().collect()
[('aaa', 2), ('abc', 3), ('bac', 1), ('bbb', 3), ('cba', 2)]

Join操作

leftOuterJoin

二つのRDDをleft outer joinして、valueに二つの要素のタプルをもつペアRDDを返す

>>> rdd1 = sc.parallelize([("Ken", 1), ("Bob", 2), ("Meg", 3)])
>>> rdd2 = sc.parallelize([("Ken", 1), ("Kaz", 3)])
>>> rdd1.leftOuterJoin(rdd2).collect()
[('Bob', (2, None)), ('Meg', (3, None)), ('Ken', (1, 1))]
rightOuterJoin

二つのRDDをright outer joinして、valueに二つの要素のタプルをもつペアRDDを返す

>>> rdd1 = sc.parallelize([("Ken", 1), ("Bob", 2), ("Meg", 3)])
>>> rdd2 = sc.parallelize([("Ken", 1), ("Kaz", 3)])
>>> rdd1.rightOuterJoin(rdd2).collect()
[('Ken', (1, 1)), ('Kaz', (3, None))]
fullOuterJoin

二つのRDDをfull outer joinして、valueに二つの要素のタプルをもつペアRDDを返す

>>> rdd1 = sc.parallelize([("Ken", 1), ("Bob", 2), ("Meg", 3)])
>>> rdd2 = sc.parallelize([("Ken", 1), ("Kaz", 3)])
>>> rdd1.fullOuterJoin(rdd2).collect()
[('Bob', (2, None)), ('Meg', (3, None)), ('Ken', (1, 1)), ('Kaz', (None, 3))]

ソート操作

sortBy

sortBy(f) fの返す値によってソートする

>>> rdd = sc.parallelize([("cba", 2), ("abc", 3), ("bac", 1), ("bbb", 
>>> rdd.sortBy(lambda (x, y): x).collect() # sortByKeyと同じ

集合操作など

intersection

intersection(rdd) 二つのRDDのintersectionを返す

union

union(rdd) 二つのRDDのunionを返す

zip

zip(rdd) 引数のrddの各要素をvlaueにしたペアRDDを返す

>>> rdd = sc.parallelize([("Ken", 2), ("Bob", 3), ("Taka", 1), ("Ken", 3), ("Bob", 2)])
>>> rdd.keys().zip(rdd.values())
[('Ken', 2), ('Bob', 3), ('Taka', 1), ('Ken', 3), ('Bob', 2)]
distinct

同じ要素を含まないRDDを返します

サンプリング操作

sample

sample(bool, frac) サンプリングしたRDDを返す。第一引数で同じ要素の重複を許すか決める。

>>> rdd = sc.parallelize([1, 2, 3, 4, 5])
>>> rdd.sample(True, 0.5).collect()
[1, 5, 5]
>>> rdd.sample(False, 0.5).collect()
[1, 3, 5]
takeSample

takeSmaple(bool, size) 固定されたサイズのサンプルをリストで返す。第一引数で同じ要素の重複を許すか決める。

>>> rdd = sc.parallelize([1, 2, 3, 4, 5])
>>> rdd.takeSample(True, 2)
[5, 5]
>>> rdd.takeSample(False, 2)
[3, 5]

デバッグ

toDebugString

toDebugString() 実行計画を返す

print(rdd.flatMap(lambda x: x.split()).map(lambda x: (x, 1)).reduceByKey(lambda x, y: x + y).toDebugString())
(1) PythonRDD[190] at RDD at PythonRDD.scala:43 []
 |  MapPartitionsRDD[189] at mapPartitions at PythonRDD.scala:374 []
 |  ShuffledRDD[188] at partitionBy at null:-1 []
 +-(1) PairwiseRDD[187] at reduceByKey at <ipython-input-114-71f5cb742e13>:1 []
    |  PythonRDD[186] at reduceByKey at <ipython-input-114-71f5cb742e13>:1 []
    |  ParallelCollectionRDD[141] at parallelize at PythonRDD.scala:423 []

永続化

persist

persist() RDDをそのまま(デフォルトではメモリに)キャッシュする。メモリだけ、メモリが無理ならディスク、ディスクだけ、などの設定が出来る(StorageLevelで指定)

>>> rdd.persist()
unpersist

unpersist() RDDの永続化を解く。永続化レベルを変える時などに使う。

>>> from pyspark import StorageLevel
>>> rdd.persist()
>>> rdd.unpersist()
>>> rdd.persist(StorageLevel.DISK_ONLY)

よくある例

随時追加していく予定

word count
>>> rdd.flatMap(lambda x: x.split())\
       .map(lambda x: (x, 1))\
       .reduceByKey(lambda x, y: x + y)\
       .take(5)

DataFrame

特に構造化データを扱うときはこちらの方が便利。

DataFrameをつくる(データの読み込み)

read.json

read.json(file) jsonからデータを読み込む

# $ cat a.json
# {"name":"Ken", "age":35}
# {"name":"Bob", "age":30, "weight":80}
# {"name":"Meg", "age":29, "weight":45}
df = sqlContext.read.json("a.json")

DataFrameを表示する

RDDと同じcollect, takeの他にshow がある

show

show(n) n行表示する(nはデフォルトで20)

>>> df.show()
+---+----+------+
|age|name|weight|
+---+----+------+
| 35| Ken|  null|
| 30| Bob|    80|
| 29| Meg|    45|
+---+----+------+

DataFrameの操作

select

select(column) stringかColumnオブジェクトを渡してselectしたDataFrameを返す。カラムを列挙して複数列取得したり、演算することもできる。

>>> df.select("age").show()
+---+
|age|
+---+
| 35|
| 30|
| 29|
+---+

# 次も同じ
>>> df.select(df.age).show() # Columnオブジェクトを渡す
>>> df.select(df["age"]).show() # Columnオブジェクトを渡す
>>> df.select(df.name, df.age).show()
+----+---+
|name|age|
+----+---+
| Ken| 35|
| Bob| 30|
| Meg| 29|
+----+---+
DataframeのColumnオブジェクト

selectで渡すColumnオブジェクトのアクセスの仕方として、Pythonでは次の2パターンが用意されている:

>>> df.age
Column<age>
>>> df["age"]
Column<age>
where/filter

filter(condition) stringの条件に合う行だけからなるDataFrameを返す。 wherefilterのaliasである。

>>> df.where(df.age >=30).show()
+---+----+------+
|age|name|weight|
+---+----+------+
| 35| Ken|  null|
| 30| Bob|    80|
+---+----+------+
sort

sort(column) 指定されたカラムでソートされたDataFrameを返す

>>> df.sort(df.age)
+---+----+------+
|age|name|weight|
+---+----+------+
| 29| Meg|    45|
| 30| Bob|    80|
| 35| Ken|  null|
+---+----+------+
limit

limit(n) 先頭n行だけに制限したDataFrameを返す

>>> df.limit(1).show()
+---+----+------+
|age|name|weight|
+---+----+------+
| 35| Ken|  null|
+---+----+------+
distinct

distinct() distinctした結果の行だけからなるDataFrameを返す

>>> df.distinct().count()
3
join

join(dataframe, on, how) howのデフォルトはinner

  • on: カラム、もしくはカラムのリスト
  • how: "inner", "outer", "left_outer", "right_outer", "leftsemi" のいずれか

DataframeからRDDへ変換

DataFrameはRDD上に構築されているので、元となるRDDを取り出すことができる

>>> print(df.rdd.collect())
[Row(age=35, name=u'Ken', weight=None),
 Row(age=30, name=u'Bob', weight=80),
 Row(age=29, name=u'Meg', weight=45)]

特定の列だけ取り出すにはRowオブジェクトの対応する属性にアクセスする

df.rdd.map(lambda row: (row.age, row.weight)).collect()
[(35, None), (30, 80), (29, 45)]

Dataframeを保存する

toJson

toJson() jsonの形でRDDに変換する。このあとsaveAsTextFileを呼べばjson形式で保存できる。

>>> df.toJSON().saveAsTextFile("b.json")
>>> df2 = sqlContext.read.json("/b.json")
+---+----+------+
|age|name|weight|
+---+----+------+
| 35| Ken|  null|
| 30| Bob|    80|
| 29| Meg|    45|
+---+----+------+

今後

Spark StreamingやMllib関連もこちらに追記していくかもしれない。

Why do not you register as a user and use Qiita more conveniently?
  1. We will deliver articles that match you
    By following users and tags, you can catch up information on technical fields that you are interested in as a whole
  2. you can read useful information later efficiently
    By "stocking" the articles you like, you can search right away
Comments
Sign up for free and join this conversation.
If you already have a Qiita account
Why do not you register as a user and use Qiita more conveniently?
You need to log in to use this function. Qiita can be used more conveniently after logging in.
You seem to be reading articles frequently this month. Qiita can be used more conveniently after logging in.
  1. We will deliver articles that match you
    By following users and tags, you can catch up information on technical fields that you are interested in as a whole
  2. you can read useful information later efficiently
    By "stocking" the articles you like, you can search right away