CUBE SUGAR CONTAINER

技術系のこと書きます。

Python: memory_profiler でプログラムのメモリ使用量を調べる

今回は memory_profiler というモジュールを使ってプログラムのメモリ使用量を調べる方法について紹介する。

pypi.python.org

このブログでは、以前に Python のプロファイラとして profile/cProfile や line_profiler について書いたことがある。 これまでに紹介したこれらのプロファイラは、主に時間計算量の調査が目的となる。 それに対して memory_profiler では、調べる対象は空間計算量となる。

blog.amedama.jp

blog.amedama.jp

使った環境は以下の通り。

$ sw_vers 
ProductName:    Mac OS X
ProductVersion: 10.12.6
BuildVersion:   16G1212
$ python --version
Python 3.6.4

下準備

まずは memory_profiler をインストールする。

$ pip install memory_profiler

スクリプトから memory_profiler を使う

まずは最も基本的な、スクリプトから memory_profiler を使う方法について。

memory_profiler では特定の関数のメモリ使用量をプロファイリングするのに @profile デコレータが使える。 例えば次のサンプルコードでは my_func() 関数を @profile デコレータでプロファイル対象としてマークしている。 関数の内容は変数の入った大きなリストを作って、それを del 文で削除するというものになる。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from memory_profiler import profile


@profile
def my_func():
    # 整数の入った大きなリストを用意する
    a = [0] * (2 * 10 ** 7)
    # 変数を削除する
    del a
    # 先ほどより少し小さなリストを用意する
    b = [0] * (10 ** 6)
    # 変数を削除する
    del b


def main():
    my_func()


if __name__ == '__main__':
    main()

上記に適当な名前をつけたら、あとは普通に Python のスクリプトとして実行するだけ。 すると、標準出力にプロファイル結果が出力される。 出力内容は、左から「プログラムの行数、その行が評価された時点でのメモリ使用量、その行が評価されたことによる使用量の増減、対応するコード」となっている。

$ python example.py 2>/dev/null
Filename: example.py

Line #    Mem usage    Increment   Line Contents
================================================
     7     36.1 MiB     36.1 MiB   @profile
     8                             def my_func():
     9                                 # 整数の入った大きなリストを用意する
    10    188.7 MiB    152.6 MiB       a = [0] * (2 * 10 ** 7)
    11                                 # 変数を削除する
    12     36.1 MiB   -152.6 MiB       del a
    13                                 # 先ほどより少し小さなリストを用意する
    14     43.8 MiB      7.6 MiB       b = [0] * (10 ** 6)
    15                                 # 変数を削除する
    16     43.8 MiB      0.0 MiB       del b

上記を見ると興味深いことが分かる。 最初の変数 a は del 文を発行することで GC が実行されたのか、メモリ使用量は減っている。 それに対し変数 b では del 文を発行してもメモリ使用量は変化していない。

IPython から memory_profiler を使う

続いては IPython からアドホックに memory_profiler を使ってみる。 おそらく、実際のプロファイリングではこの方法を使うことが多いだろう。

まずは IPython をインストールしておく。

$ pip install ipython

先ほどと、ほぼ同じ内容のサンプルコードを用意する。 違いは my_func() 関数に @profile デコレータが付与されていないところだ。

#!/usr/bin/env python
# -*- coding: utf-8 -*-


def my_func():
    a = [0] * (2 * 10 ** 7)
    del a
    b = [0] * (10 ** 6)
    del b


def main():
    my_func()


if __name__ == '__main__':
    main()

上記で用意した example.py を IPython に読み込ませながら起動する。

$ ipython -i example.py

ちなみに、上記で起動と同時にモジュールを読み込ませているのは手順を省くためだけ。 単独で IPython を起動した後に my_func 関数をインポートしても、もちろん構わない。

$ ipython
...
In [1]: from example import my_func

memory_profiler の IPython 拡張を読み込む。

In [1]: %load_ext memory_profiler

あとは %mprun マジックコマンド経由で my_func() 関数を実行する。 これで、先ほどスクリプトから実行したのと同じ内容が得られる。

In [2]: %mprun -f my_func my_func()
Filename: /Users/amedama/Documents/temporary/example.py

Line #    Mem usage    Increment   Line Contents
================================================
     5     49.2 MiB     49.2 MiB   def my_func():
     6    201.8 MiB    152.6 MiB       a = [0] * (2 * 10 ** 7)
     7     49.2 MiB   -152.6 MiB       del a
     8     49.2 MiB      0.0 MiB       b = [0] * (10 ** 6)
     9     49.2 MiB      0.0 MiB       del b

処理内容が一行で収まるときは %memit マジックコマンドも便利だ。

In [3]: %memit list(range(10 ** 6))
peak memory: 86.92 MiB, increment: 27.92 MiB

In [4]: %memit list(range(10 ** 7))
peak memory: 437.59 MiB, increment: 378.59 MiB

プロファイル結果を matplotlib で折れ線グラフにプロットする

memory_profiler は matplotlib と連携してプロファイル結果をプロットする機能もある。

そのために、まずは matplotlib をインストールしておこう。

$ pip install matplotlib

サンプルコードを mprof run コマンド経由で実行する。

$ mprof run example.py
mprof: Sampling memory every 0.1s
running as a Python program...

実行が完了したら mprof plot コマンドでプロットされた結果を確認する。

$ mprof plot
Using last profile data.

こんな感じで結果が確認できる。 f:id:momijiame:20180204001410p:plain

ちなみにプロファイル結果は履歴が残る。 履歴は mprof list コマンドで確認できる。

$ mprof list
0 mprofile_20180203230956.dat 23:09:56 03/02/2018

インデックスを指定すれば過去の実行結果のプロットが確認できる。

$ mprof plot 0

履歴を削除するにはインデックスを指定して mprof rm コマンドを実行する。

$ mprof rm 0

あるいは、全ての履歴を削除したいなら mprof clean コマンドを使っても構わない。

$ mprof clean

めでたしめでたし。

PySpark の UDF (User Defined Function) を試す

今回は PySpark の UDF (User Defined Function) 機能を使ってみる。 UDF というのはユーザが定義した関数を使って Spark クラスタで分散処理をするための機能になっている。 柔軟に処理を記述できるメリットがある一方で、パフォーマンスには劣るというデメリットもある。 この特性は、ユーザが定義した処理をワーカーに配布した上で Python インタプリタに解釈させる特性に由来している。 今回は、そんな UDF を DataFrame API と Spark SQL という二つの API を通して使ってみることにした。

使った環境は次の通り。 クラスタは YARN で管理している。

$ pyspark --version
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 2.2.1
      /_/
                        
Using Scala version 2.11.8, OpenJDK 64-Bit Server VM, 1.8.0_161
Branch 
Compiled by user felixcheung on 2017-11-24T23:19:45Z
Revision 
Url 
Type --help for more information.

下準備

まずは UDF で処理する DataFrame オブジェクトを用意しよう。

最初に、ユーザの情報を模した RDD を定義する。

>>> rdd = sc.parallelize([
...   ('Alice', 20),
...   ('Bob', 25),
...   ('Carol', 30),
... ])

続いて、上記 RDD のスキーマを定義する。

>>> from pyspark.sql.types import StructType
>>> from pyspark.sql.types import StructField
>>> from pyspark.sql.types import StringType
>>> from pyspark.sql.types import IntegerType
>>> schema = StructType([
...   StructField('name', StringType(), False),
...   StructField('age', IntegerType(), False),
... ])

RDD とスキーマから DataFrame を生成する。

>>> df = spark.createDataFrame(rdd, schema)

また、後ほど Spark SQL から操作する場合のことを考えて、この DataFrame をテーブルとして扱えるようにしておこう。

>>> df.registerTempTable('users')

UDF となる関数を定義する

続いて UDF にする関数を定義する。 これは、何の変哲もない Python の関数でしかない。 引数としてカラムの値を受け取って、何らかの加工したカラムの値を返すことになる。

>>> def double(column):
...     return column * 2
... 

今回のサンプルコードでは、受け取った引数を 2 倍にして返すという単純なものにした。

DataFrame API から UDF を使う

まずは DataFrame API で UDF を使う方法から。 これには pyspark.sql.functions.udf() という関数を使う。

>>> from pyspark.sql.functions import udf

上記の udf() 関数を使って、先ほど定義した関数をラップする。 このラップした udf_double() 関数が UDF として動作する。

>>> udf_double = udf(double)

後は pyspark.sql.functions にあるような関数と同じような使い勝手で UDF が使える。 例えばサンプルデータに UDF を使って age カラムの値を 2 倍してみよう。

>>> df.select(udf_double('age')).show()
+-----------+
|double(age)|
+-----------+
|         40|
|         50|
|         60|
+-----------+

上手くいったようだ。

上記だと、表示がちょっと分かりにくいかもしれないので他のカラムも同時に出力してみる。 カラムには alias() メソッドを使って名前が付けられる。

>>> df.select('*', udf_double('age').alias('doubled_age')).show()
+-----+---+-----------+
| name|age|doubled_age|
+-----+---+-----------+
|Alice| 20|         40|
|  Bob| 25|         50|
|Carol| 30|         60|
+-----+---+-----------+

Spark SQL から UDF を使う

続いては Spark SQL から UDF を使ってみる。

それには、まず spark.udf.register() 関数を使って定義した関数を UDF として登録する。

>>> spark.udf.register('udf_double', double)

あとは Spark SQL で処理する SQL 文の中で一般的な関数のように使うことができる。

>>> spark.sql('''
... SELECT
...   *,
...   udf_double(age) AS doubled_age
... FROM users
... ''').show()
+-----+---+-----------+
| name|age|doubled_age|
+-----+---+-----------+
|Alice| 20|         40|
|  Bob| 25|         50|
|Carol| 30|         60|
+-----+---+-----------+

ばっちり。

めでたしめでたし。

入門 PySpark ―PythonとJupyterで活用するSpark 2エコシステム

入門 PySpark ―PythonとJupyterで活用するSpark 2エコシステム

Sparkによる実践データ解析 ―大規模データのための機械学習事例集

Sparkによる実践データ解析 ―大規模データのための機械学習事例集

初めてのSpark

初めてのSpark

  • 作者: Holden Karau,Andy Konwinski,Patrick Wendell,Matei Zaharia,Sky株式会社玉川竜司
  • 出版社/メーカー: オライリージャパン
  • 発売日: 2015/08/22
  • メディア: 大型本
  • この商品を含むブログ (4件) を見る

Docker コンテナの動作に必要な設定を起動時に渡す

今回は Docker コンテナを起動するタイミングで、コンテナの動作に必要な設定を受け渡す方法について書く。 やり方としては、大まかに分けて「環境変数を通して渡す」と「コマンドライン引数を通して渡す」という二つがある。 どちらの場合も docker run で実行するコマンドの中に設定を含めることになる。

使った環境は次の通り。

$ sw_vers
ProductName:    Mac OS X
ProductVersion: 10.12.6
BuildVersion:   16G1212
$ docker version
Client:
 Version:   18.01.0-ce
 API version:   1.35
 Go version:    go1.9.2
 Git commit:    03596f5
 Built: unknown-buildtime
 OS/Arch:   darwin/amd64
 Experimental:  false
 Orchestrator:  swarm

Server:
 Engine:
  Version:  18.01.0-ce
  API version:  1.35 (minimum version 1.12)
  Go version:   go1.9.2
  Git commit:   03596f5
  Built:    Wed Jan 10 20:13:12 2018
  OS/Arch:  linux/amd64
  Experimental: false

環境変数を通して渡す

まずは環境変数を通して渡す一般的なやり方から。

最初は動作確認のために、起動時に環境変数の一覧を表示する Docker イメージを作ることにする。 その Dockerfile が次の通り。

$ cat << 'EOF' > Dockerfile
FROM alpine

CMD env
EOF

Alpine Linux をベースイメージにして、起動時に実行するコマンドを CMD 命令で指定している。 env コマンドは環境変数を一覧で表示する。 これ以上ないくらいシンプル。

上記の Docker ファイルをビルドして Docker イメージを作る。

$ docker build -t example/env .
...
Successfully tagged example/env:latest

上記でビルドした Docker イメージからコンテナを起動してみよう。 すると、コンテナ内で設定されている環境変数が出力される。

$ docker run -t example/env
HOSTNAME=4df5d62c61a1
SHLVL=1
HOME=/root
TERM=xterm
PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
PWD=/

続いてコンテナに環境変数を設定してみる。 これには docker run コマンドで -e (--env) オプションを指定する。 例えば、よくありがちなバインドするアドレスやポートっぽい値を設定してみよう。

$ docker run -e BIND_ADDRESS=127.0.0.1 -e BIND_PORT=8080 -t example/env
BIND_ADDRESS=127.0.0.1
HOSTNAME=2fb165c97c2a
SHLVL=1
HOME=/root
BIND_PORT=8080
TERM=xterm
PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
PWD=/

ちゃんと環境変数に BIND_ADDRESSBIND_PORT が設定されたことが分かる。

ちなみに、環境変数はファイル経由で渡すこともできる。 渡す変数の数が多いときは、こちらを使った方が良い。

まずは環境変数を羅列したファイルを用意する。

$ cat << 'EOF' > envfile.txt
BIND_ADDRESS=127.0.0.1
BIND_PORT=8080
EOF

あとはコンテナを起動するときに --env-file オプションでファイルを指定する。

$ docker run --env-file envfile.txt -t example/env
HOSTNAME=4ed1b4a9a45c
BIND_ADDRESS=127.0.0.1
SHLVL=1
BIND_PORT=8080
HOME=/root
TERM=xterm
PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
PWD=/

デフォルト値を設定する (Dockerfile)

上記でアプリケーションの設定として環境変数を使う下地が整った。 ただ、設定には一般的にデフォルト値がほしくなる。

そんなときは Dockerfile で ENV 命令を使うことができる。 先ほどの Dockerfile に ENV 命令を加えたものを用意しよう。

$ cat << 'EOF' > Dockerfile
FROM alpine

ENV BIND_ADDRESS 127.0.0.1
ENV BIND_PORT 8080

CMD env
EOF

上記のファイルをビルドする。

$ docker build -t example/env .
...
Successfully tagged example/env:latest

ビルドしたイメージからコンテナを起動してみよう。 特に -e (--env) オプションを使わなくても環境変数が出力されていることが分かる。

$ docker run -t example/env
BIND_ADDRESS=127.0.0.1
HOSTNAME=4c795ff84a32
SHLVL=1
HOME=/root
BIND_PORT=8080
TERM=xterm
PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
PWD=/

上記の値は -e (--env) オプションを指定することで上書きできる。

$ docker run -e BIND_ADDRESS=0.0.0.0 -e BIND_PORT=80 -t example/env
BIND_ADDRESS=0.0.0.0
HOSTNAME=de360066a866
SHLVL=1
HOME=/root
BIND_PORT=80
TERM=xterm
PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
PWD=/

デフォルト値を設定する (シェルスクリプト)

環境変数のデフォルト値を設定するには Dockerfile で ENV 命令を使う以外のやり方もある。 具体的には、コンテナが起動するタイミングでシェルスクリプトを実行して、その中で環境変数を操作する。

以下の Dockerfile では docker-entrypoint.sh というファイルをイメージに転送している。 そして ENTRYPOINT 命令を使うことで、コンテナの起動時にそのシェルスクリプトを実行するように設定されている。

$ cat << 'EOF' > Dockerfile 
FROM alpine

COPY docker-entrypoint.sh /usr/local/bin

ENTRYPOINT ["docker-entrypoint.sh"]
EOF

続いて、上記の Dockerfile で使っている docker-entrypoint.sh を用意する。 やっていることは単純で、環境変数がないときはデフォルト値を扱うように ${環境変数名:-デフォルト値} という記法を使うだけ。

$ cat << 'EOF' > docker-entrypoint.sh 
#!/bin/sh

echo ${BIND_ADDRESS:-127.0.0.1}
echo ${BIND_PORT:-8080}
EOF
$ chmod +x docker-entrypoint.sh 

上記で用意したファイル群から Docker イメージをビルドする。

$ docker build -t example/env .
...
Successfully tagged example/env:latest

イメージからコンテナを起動する。 特にオプションを指定のないときはシェルスクリプトで指定したデフォルト値が使われる。 また、先ほどと同じように -e (--env) オプションを指定することで値を上書きできる。

$ docker run -t example/env
127.0.0.1
8080
$ docker run -e BIND_ADDRESS=0.0.0.0 -e BIND_PORT=80 -t example/env
0.0.0.0
80

もちろんシェルスクリプトの中では受け渡された環境変数の内容をバリデーションしたりもできる。

コマンドライン引数を通して渡す

続いてはもう一つのやり方、コマンドライン引数を使う方法について。 このやり方は、基本的には前述したシェルスクリプトを使う方法の応用になっている。

まずは、先ほどと同じように起動時にシェルスクリプトを実行するような Docker ファイルを用意する。 ここでシェルスクリプトを実行するのに ENTRYPOINT 命令を使っているのがポイントになる。 これが CMD 命令だと上手くいかない。

$ cat << 'EOF' > Dockerfile 
FROM alpine

COPY docker-entrypoint.sh /usr/local/bin

ENTRYPOINT ["docker-entrypoint.sh"]
EOF

実行されるシェルスクリプトは、全てのコマンドライン引数を参照できる $@ 変数を echo コマンドで出力する。

$ cat << 'EOF' > docker-entrypoint.sh 
#!/bin/sh

echo $@
EOF
$ chmod +x docker-entrypoint.sh

上記をビルドしよう。

$ docker build -t example/opt .   
...
Successfully tagged example/opt:latest

上記のイメージからコンテナを起動するタイミングで、普段なら起動するコマンドを渡すところに適当な文字列を入れてみよう。

$ docker run -t example/opt foo bar baz                          
foo bar baz

すると、入力した文字列がそのまま出力された。 これはシェルスクリプトがコマンドライン引数を出力するようにした echo $@ による結果となる。 つまり、コマンドライン引数をシェルスクリプトに渡すことができたというわけ。

コマンドライン引数を解析する (getopts)

シェルスクリプトにコマンドライン引数さえ渡せてしまえば、あとはどうとでもなる。 一例として、ここでは getopts を使って引数を解析してみることにした。

起動するシェルスクリプトで getopts を使って受け取った引数を解析する。 そして、最終的には解析した変数を出力している。 ここでは -a オプションで渡した値が BIND_ADDRESS に、-p オプションで渡したあたいが BIND_PORT に格納される。

$ cat << 'EOF' > docker-entrypoint.sh 
#!/bin/sh

usage() {
  echo "Usage: $0 [-a bind-address] [-p bind-port]" 1>&2
  exit 1
}

while getopts a:p:h OPT
do
  case $OPT in
    a)  BIND_ADDRESS=$OPTARG
        ;;
    p)  BIND_PORT=$OPTARG
        ;;
    h)  usage
        ;;
  esac
done

echo ${BIND_ADDRESS:-127.0.0.1}
echo ${BIND_PORT:-8080}
EOF
$ chmod +x docker-entrypoint.sh 

上記を元にイメージをビルドしよう。

$ docker build -t example/opt .             
...
Successfully tagged example/opt:latest

そしてイメージからコンテナを起動する。 特に何も指定しないときはデフォルト値が表示され、オプションをコマンドライン引数で指定したときはその値が表示される。

$ docker run -t example/opt            
127.0.0.1
8080
$ docker run -t example/opt -a 0.0.0.0 -p 80
0.0.0.0
80

これだと、例えば -h を渡したときは usage を表示して終了みたいなことも簡単にできる。

$ docker run -t example/opt -h
Usage: /usr/local/bin/docker-entrypoint.sh [-a bind-address] [-p bind-port]

ちなみに上記のやり方を取るとデバッグしたいときにどうするんだって話になる。 起動時のパラメータの最後に bin/bash とか付けるだけではシェルスクリプトの起動が上書きできないので。 そんなときは --entrypoint オプションを使えば ENTRYPOINT 命令の内容を上書きできる。

$ docker run --entrypoint sh -it example/opt
/ # uname -a
Linux a6c3524f68a5 4.4.111-boot2docker #1 SMP Thu Jan 11 16:25:31 UTC 2018 x86_64 Linux

ばっちり。

Docker のホストとコンテナ間でファイルをやり取りする

Docker ホストとコンテナの間でファイルをやり取りするのって以前はかなり面倒だったと思う。 そんな記憶も今は昔、専用のコマンドが用意されてだいぶ簡単になっているようだ。

docker cp | Docker Documentation

使った環境は次の通り。

$ sw_vers
ProductName:    Mac OS X
ProductVersion: 10.12.6
BuildVersion:   16G1212
$ docker version
Client:
 Version:   18.01.0-ce
 API version:   1.35
 Go version:    go1.9.2
 Git commit:    03596f5
 Built: unknown-buildtime
 OS/Arch:   darwin/amd64
 Experimental:  false
 Orchestrator:  swarm

Server:
 Engine:
  Version:  18.01.0-ce
  API version:  1.35 (minimum version 1.12)
  Go version:   go1.9.2
  Git commit:   03596f5
  Built:    Wed Jan 10 20:13:12 2018
  OS/Arch:  linux/amd64
  Experimental: false

下準備

まずはファイルをやり取りするための Docker コンテナを起動しておく。

$ docker run -it alpine /bin/sh

Alpine Linux の Docker コンテナを起動してシェルに入った。

/ # uname -a
Linux 09a735dae425 4.4.111-boot2docker #1 SMP Thu Jan 11 16:25:31 UTC 2018 x86_64 Linux

起動したコンテナの ID を確認しておく。

$ docker ps
CONTAINER ID        IMAGE               COMMAND             CREATED             STATUS              PORTS               NAMES
09a735dae425        alpine              "/bin/sh"           12 seconds ago      Up 14 seconds                           stoic_mclean

ホストからコンテナにファイルをコピーする

まずは Docker ホストからコンテナにファイルをコピーする方から。

Docker ホスト側で、まずはコピーするファイルを用意する

$ echo "Hello, World" > greeting.txt

あとは docker cp コマンドを使ってファイルを転送するだけ。 コマンドのパラメータは scp コマンドのイメージに近い。 送信元を第一引数に、送信先を第二引数に指定する。 送信先にはコンテナ ID (もしくは名前) と、コロンで区切って転送先のパスを指定する。

$ docker cp greeting.txt 09a735dae425:/tmp

docker exec コマンドを使ってコンテナ上で cat コマンドを実行してファイルの状態を確認しよう。

$ docker exec 09a735dae425 cat /tmp/greeting.txt
Hello, World

ちゃんとファイルが送られたことが分かる。

コンテナからホストにファイルをコピーする

続いてはコンテナからホストにファイルをコピーするパターン。 この場合も、最終的には先ほどと同じように docker cp コマンドを使う。

まずはコンテナの中でコピーするファイルを用意しよう。 分かりやすいように、先ほど送り込まれたファイルを編集して使う。

/ # echo "Hello, Docker" > /tmp/greeting.txt

あとは Docker ホスト側で docker cp コマンドを使ってコンテナ上のファイルをホストに転送してくる。 今回は送信元となる第一引数の方にコンテナとファイルの絶対パスを指定する。

$ docker cp 09a735dae425:/tmp/greeting.txt .

コマンドを実行すると、ちゃんとコンテナにあったファイルが手元にコピーされたことが分かる。

$ cat greeting.txt 
Hello, Docker

いじょう。

めでたしめでたし。

PySpark: Jupyter Notebook からローカルの PySpark ドライバを操作する

今回はローカルで動作している PySpark のドライバプログラムを Jupyter Notebook から操作する方法について。 ようするに Jupyter Notebook と PySpark (のドライバ) が同じマシン上で動く場合ということ。 この場合の設定内容はとてもシンプルで、環境変数を書き換えるだけで使えるようになる。

これがもし Jupyter Notebook と PySpark のホストが分かれていると、もうちょっと複雑になる。 具体的には Apache Ivy と SparkMagic というソフトウェアを連携させないといけない。 それについては今後改めて書くつもり。

使った環境は次の通り。

$ cat /etc/redhat-release
CentOS Linux release 7.4.1708 (Core)
$ uname -r
3.10.0-693.11.1.el7.x86_64
$ pyspark --version
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 2.2.1
      /_/
                        
Using Scala version 2.11.8, OpenJDK 64-Bit Server VM, 1.8.0_161
Branch 
Compiled by user felixcheung on 2017-11-24T23:19:45Z
Revision 
Url 
Type --help for more information.
$ python3 -V
Python 3.6.4

ドライバのホストに Jupyter Notebook をインストールする

まずは PySpark のドライバを動かしているホストに Jupyter Notebook をインストールする。 どうせ後から必要になるので pandas とかも入れておくと良いかも。

$ sudo pip3 install jupyter pandas

PySpark が Jupyter Notebook を起動するように環境変数を設定する

続いては PySpark を使うときインタプリタとして Jupyter Notebook を使うように環境変数を設定する。

$ cat << 'EOF' >> ~/.bashrc
export PYSPARK_PYTHON=/usr/bin/python3
export PYSPARK_DRIVER_PYTHON=jupyter
export PYSPARK_DRIVER_PYTHON_OPTS='notebook' pyspark
EOF
$ source ~/.bashrc

後は通常通り PySpark を起動すれば、同時に Jupyter Notebook が使えるようになる。 この環境では YARN を使って Spark クラスタを構築した。 そのため --masteryarn を指定している。

$ pyspark --master yarn

あとはブラウザで Jupyter Notebook が動作している場所を開くだけ。

$ open http://localhost:8888

PySpark のドライバがリモートで動作している場合

作業しているホストがローカルホストでないときは PYSPARK_DRIVER_PYTHON_OPTS で Listen するアドレスを指定する。 任意のアドレスで Listen するときは、次のように --ip0.0.0.0 を指定する。

$ cat << 'EOF' >> ~/.bashrc
export PYSPARK_PYTHON=/usr/bin/python3.6
export PYSPARK_DRIVER_PYTHON=jupyter
export PYSPARK_DRIVER_PYTHON_OPTS='notebook --ip=0.0.0.0' pyspark
EOF

先ほど紹介したのと同じように起動する。

$ pyspark --master yarn

起動するとコンソールにトークンを含む URL が表示されるはず。

表示された URL を、必要に応じて IP アドレス部分を書き換えたらブラウザで開くだけ。

$ open http://192.168.33.10:8888/?token=XXXXX...

動作確認

Jupyter Notebook の画面が開いたら新しいノートブックを作成して、ちゃんと PySpark が動くか確認しよう。

例えば SparkContext がちゃんと使えるか、とか。

In:
sc

Out:
SparkContext

Spark UI

Version
v2.2.1
Master
yarn
AppName
PySparkShell

分散処理の部分がちゃんと動くかを確かめるためにワードカウントしてみたりね。

In:
rdd = sc.parallelize(['A', 'B', 'C', 'A'])
keyvalues = rdd.map(lambda x: (x, 1))
counts = keyvalues.reduceByKey(lambda a, b: a + b)
counts.collect()

Out:
[('A', 2), ('B', 1), ('C', 1)]

いじょう。

入門 PySpark ―PythonとJupyterで活用するSpark 2エコシステム

入門 PySpark ―PythonとJupyterで活用するSpark 2エコシステム

PySpark: 時刻と文字列を相互に変換する (DataFrame / Spark SQL)

今回は Apache Spark のインターフェースの一つである PySpark で時刻と文字列を相互に変換する方法について扱う。 PySpark にはいくつかの API があるけど、その中でも DataFrame と Spark SQL を使った方法について紹介する。

使った環境は次の通り。

$ pyspark --version
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 2.2.1
      /_/
                        
Using Scala version 2.11.8, OpenJDK 64-Bit Server VM, 1.8.0_161
Branch 
Compiled by user felixcheung on 2017-11-24T23:19:45Z
Revision 
Url 
Type --help for more information.

下準備

まずは時刻情報を文字列で表した RDD オブジェクトを用意する。 Apache Spark では主に RDD と DataFrame という二つのオブジェクトを中心にデータを操作する。 その中でも RDD はスキーマレスなオブジェクトになっている。

>>> rdd = sc.parallelize([
...   ('01/Jan/2016 10:13:16',),
...   ('02/Feb/2017 11:14:17',),
...   ('03/Mar/2018 12:15:18',)
... ])

時刻の形式は英語圏にありがちなフォーマットにした。

RDD を DataFrame に変換するためにスキーマ情報を定義する。

>>> from pyspark.sql.types import StructType
>>> from pyspark.sql.types import StructField
>>> from pyspark.sql.types import StringType
>>> schema = StructType([
...   StructField('dt_str', StringType(), False),
... ])

RDD とスキーマを元に DataFrame を生成する。

>>> df = spark.createDataFrame(rdd, schema)
>>> df.show()
+--------------------+
|              dt_str|
+--------------------+
|01/Jan/2016 10:13:16|
|02/Feb/2017 11:14:17|
|03/Mar/2018 12:15:18|
+--------------------+

これで準備ができた。

string から timestamp への変換 (Spark 2.2 ~)

Apache Spark 2.2 以降では、文字列の時刻情報をタイムスタンプに変換する関数として to_timestamp() が用意されている。 この関数は例えば以下のようにして使う。

>>> from pyspark.sql.functions import to_timestamp
>>> df.select(to_timestamp(df.dt_str, 'dd/MMM/yyyy HH:mm:ss').alias('parsed_dt')).show()
+-------------------+
|          parsed_dt|
+-------------------+
|2016-01-01 10:13:16|
|2017-02-02 11:14:17|
|2018-03-03 12:15:18|
+-------------------+

変換した DataFrame の型を調べると、ちゃんと timestamp 型になっていることが分かる。

>>> df.select(to_timestamp(df.dt_str, 'dd/MMM/yyyy HH:mm:ss').alias('parsed_dt')).dtypes
[('parsed_dt', 'timestamp')]

string から timestamp への変換 (Spark 1.5 ~)

先ほど使った API はかなり新しいので使えない環境もあるかと思う。 そこで Apache Spark 1.5 以降であれば次のように unix_timestamp() 関数と cast() メソッドを組み合わせると良い。 ようするに、一旦 UNIX タイムにした上でそれを timestamp として解釈させるということ。

>>> from pyspark.sql.functions import unix_timestamp
>>> df.select(unix_timestamp(df.dt_str, 'dd/MMM/yyyy HH:mm:ss').cast('timestamp').alias('parsed_dt')).show()
+-------------------+
|          parsed_dt|
+-------------------+
|2016-01-01 10:13:16|
|2017-02-02 11:14:17|
|2018-03-03 12:15:18|
+-------------------+

変換した後の DataFrame の型を調べると、ちゃんと timestamp 型が使われていることが分かる。

>>> df.select(unix_timestamp(df.dt_str, 'dd/MMM/yyyy HH:mm:ss').cast('timestamp').alias('parsed_dt')).dtypes
[('parsed_dt', 'timestamp')]

timestamp 型に変換すると、次のように filter() メソッドで範囲指定なんかができる。

>>> parsed_df = df.select(to_timestamp(df.dt_str, 'dd/MMM/yyyy HH:mm:ss').alias('parsed_dt'))
>>> parsed_df.filter('parsed_dt > "2017"').show()
+-------------------+
|          parsed_dt|
+-------------------+
|2017-02-02 11:14:17|
|2018-03-03 12:15:18|
+-------------------+

string から timestamp への変換 (Spark SQL)

ちなみに上記の変換は DataFrame API を使う以外に Spark SQL を使うこともできる。 Spark SQL を使うと、文字通り SQL を通して各種データを操作できる。

Spark SQL を使うには、まず DataFrame を registerTempTable() メソッドを使ってテーブルとして扱えるようにする。

>>> df.registerTempTable('datetimes')

すると SparkSession オブジェクトの sql() メソッドで、上記で登録したテーブルを SQL から触れるようになる。

>>> spark.sql('''
... SELECT
...   CAST(from_unixtime(unix_timestamp(dt_str, 'dd/MMM/yyyy HH:mm:ss')) AS timestamp) AS parsed_dt
... FROM datetimes
... ''').show()
+-------------------+
|          parsed_dt|
+-------------------+
|2016-01-01 10:13:16|
|2017-02-02 11:14:17|
|2018-03-03 12:15:18|
+-------------------+

timestamp を string に変換する (Spark 1.5 ~)

続いては、これまでとは逆にタイムスタンプを文字列に変換してみよう。

タイムスタンプから文字列の変換には Apache Spark 1.5 以降であれば date_format() 関数が使える。

>>> from pyspark.sql.functions import date_format
>>> parsed_df.select(date_format(parsed_df.parsed_dt, 'dd/MMM/yyyy HH:mm:ss').alias('dt_str')).show()
+--------------------+
|              dt_str|
+--------------------+
|01/Jan/2016 10:13:16|
|02/Feb/2017 11:14:17|
|03/Mar/2018 12:15:18|
+--------------------+

変換後の型情報を確認すると、ちゃんと文字列になっていることが分かる。

>>> parsed_df.select(date_format(parsed_df.parsed_dt, 'dd/MMM/yyyy HH:mm:ss').alias('dt_str')).dtypes
[('dt_str', 'string')]

ちなみに、凝ったフォーマットが不要であれば以下のように文字列にキャストしてしまうだけでも事足りる。

>>> parsed_df.select(parsed_df.parsed_dt.cast('string').alias('dt_str')).show()
+-------------------+
|             dt_str|
+-------------------+
|2016-01-01 10:13:16|
|2017-02-02 11:14:17|
|2018-03-03 12:15:18|
+-------------------+

この場合でも、ちゃんと文字列に変換されている。

>>> parsed_df.select(parsed_df.parsed_dt.cast('string').alias('dt_str')).dtypes
[('dt_str', 'string')]

timestamp を string に変換する (Spark SQL)

上記の操作は、もちろん Spark SQL を使ってもできる。 先ほどと同じように、まずは DataFrame を registerTempTable() メソッドでテーブルとして扱えるようにする。

>>> parsed_df.registerTempTable('datetimes')

あとは SparkSession オブジェクト経由で、型を変換する SELECT 文を書けばいいだけ。

>>> spark.sql('''
... SELECT
...   date_format(parsed_dt, 'dd/MMM/yyyy HH:mm:ss') AS dt_str
... FROM datetimes
... ''').show()
+--------------------+
|              dt_str|
+--------------------+
|01/Jan/2016 10:13:16|
|02/Feb/2017 11:14:17|
|03/Mar/2018 12:15:18|
+--------------------+

いじょう。

めでたしめでたし。

入門 PySpark ―PythonとJupyterで活用するSpark 2エコシステム

入門 PySpark ―PythonとJupyterで活用するSpark 2エコシステム

Apache Hive を使ったテーブルのサンプリング

Apache Hive では、大規模なデータセットに対してクエリを実行すると完了までに長い時間がかかる。 そこで、全体から一部を抽出した標本に対してクエリを実行する場合がある。 今回は、その標本を抽出する方法 (サンプリング) について扱う。

使った環境は次の通り。

$ cat /etc/redhat-release 
CentOS Linux release 7.4.1708 (Core)
$ uname -r
3.10.0-693.5.2.el7.x86_64
$ hadoop version
Hadoop 2.8.3
Subversion https://git-wip-us.apache.org/repos/asf/hadoop.git -r b3fe56402d908019d99af1f1f4fc65cb1d1436a2
Compiled by jdu on 2017-12-05T03:43Z
Compiled with protoc 2.5.0
From source with checksum 9ff4856d824e983fa510d3f843e3f19d
This command was run using /home/vagrant/hadoop-2.8.3/share/hadoop/common/hadoop-common-2.8.3.jar
$ hive --version
Hive 2.3.2
Git git://stakiar-MBP.local/Users/stakiar/Desktop/scratch-space/apache-hive -r 857a9fd8ad725a53bd95c1b2d6612f9b1155f44d
Compiled by stakiar on Thu Nov 9 09:11:39 PST 2017
From source with checksum dc38920061a4eb32c4d15ebd5429ac8a

下準備

まずは例となるクエリを実行するテーブルを用意しておこう。 これのテーブルは、整数を格納するカラムを一つだけ持っている。

hive> CREATE TABLE numbers (
    >   n INTEGER
    > );
OK
Time taken: 0.066 seconds

上記のテーブルに対してレコードを追加していく。 ここで注意すべきなのは INSERT 文を個別に発行すること。 詳しくは後述するものの、クエリを一つのまとめてしまうとブロックサンプリングという方法を使ったときに上手く動作しない。

hive> INSERT INTO TABLE numbers VALUES (0);
...
hive> INSERT INTO TABLE numbers VALUES (1);
...
hive> INSERT INTO TABLE numbers VALUES (9);
...
OK
_col0
Time taken: 2.384 seconds

テーブルが以下のような状況になっていることを確認する。

hive> SELECT * FROM numbers;;
OK
numbers.n
0
1
2
3
4
5
6
7
8
9
Time taken: 0.139 seconds, Fetched: 10 row(s)

これで、ひとまず準備ができた。

ランダムなサンプリング

Apache Hive でテーブルの一部をサンプリングするには TABLESAMPLE という構文を使う。 この構文にはいくつかの使い方があるものの、基本は次のようなクエリとなる。

hive> SELECT *
    > from numbers
    > TABLESAMPLE(BUCKET 1 OUT OF 10 ON rand());
OK
numbers.n
6
Time taken: 0.048 seconds, Fetched: 1 row(s)

上記のクエリでは、まず格納されているそれぞれのレコードに rand() 関数でランダムな値を割り振っている。 そして、そのランダムな値をハッシュ化して、結果を 10 個のバケットに振り分けていく。 振り分けられたバケットのうち 1 番目を出力する、というのが上記のクエリの意味となる。

乱数をハッシュ化して 10 個のバケットに割り振っているため、それぞれのバケットには概ね 1 つずつレコードが入ることが期待される。 しかし、もちろん偏ることもあるので次のように 2 つ以上入っていたり、反対に全く入らないこともある。

hive> SELECT *
    > from numbers
    > TABLESAMPLE(BUCKET 1 OUT OF 10 ON rand());
OK
numbers.n
5
6
Time taken: 0.047 seconds, Fetched: 2 row(s)

これはもちろんバケットの数を減らしたり増やした場合にも同じことがいえる。

hive> SELECT *
    > from numbers
    > TABLESAMPLE(BUCKET 1 OUT OF 5 ON rand());
OK
numbers.n
3
6
Time taken: 0.046 seconds, Fetched: 2 row(s)
hive> SELECT *
    > from numbers
    > TABLESAMPLE(BUCKET 1 OUT OF 5 ON rand());
OK
numbers.n
2
5
6
Time taken: 0.062 seconds, Fetched: 3 row(s)

特定のカラムをハッシュ化に用いる

先ほどの例ではハッシュ化に用いる値に rand() 関数が返すランダムな値を使った。 しかし、これにはテーブルに存在する特定のカラムを用いることもできる。

例えば numbers テーブルの n カラムをハッシュ化に使ってみよう。

hive> SELECT *
    > from numbers
    > TABLESAMPLE(BUCKET 1 OUT OF 10 ON n);
OK
numbers.n
0
Time taken: 0.065 seconds, Fetched: 1 row(s)

特定のカラムをハッシュ化に使う場合、値は実行ごとに変化することがないため毎回同じ内容が得られる。

hive> SELECT *
    > from numbers
    > TABLESAMPLE(BUCKET 1 OUT OF 10 ON n);
OK
numbers.n
0
Time taken: 0.068 seconds, Fetched: 1 row(s)

取得したい内容を変更するには、選択するバケットを変えるしかない。

hive> SELECT *
    > from numbers
    > TABLESAMPLE(BUCKET 2 OUT OF 10 ON n);
OK
numbers.n
1
Time taken: 0.057 seconds, Fetched: 1 row(s)
hive> SELECT *
    > from numbers
    > TABLESAMPLE(BUCKET 3 OUT OF 10 ON n);
OK
numbers.n
2
Time taken: 0.053 seconds, Fetched: 1 row(s)

ブロックサンプリング

TABLESAMPLE には特定の割合をサンプリングするよう指定する方法もある。 これはブロックサンプリングと呼ばれるやり方で、その名の通りテーブルを構成するブロック単位でサンプリングする。 下準備で INSERT を一つのクエリにまとめなかったのはこのためだった。 一つのクエリにまとめてしまうと、レコードが全て一つのブロックに格納されてしまうため、この機能が上手く動作しない。

ブロックサンプリングでは TABLESAMPLE に百分率で割合を指定する。

hive> SELECT *
    > FROM numbers
    > TABLESAMPLE(10.0 PERCENT);
OK
numbers.n
0
Time taken: 0.026 seconds, Fetched: 1 row(s)
hive> SELECT *
    > FROM numbers
    > TABLESAMPLE(20.0 PERCENT);
OK
numbers.n
0
1
Time taken: 0.029 seconds, Fetched: 2 row(s)

ただし、この実行結果についても、そのままでは毎回同じ内容が得られる。

hive> SELECT *
    > FROM numbers
    > TABLESAMPLE(20.0 PERCENT);
OK
numbers.n
0
1
Time taken: 0.037 seconds, Fetched: 2 row(s)

得られる内容を変更したいときは、明示的に hive.sample.seednumber を変更してやる必要がある。

hive> set hive.sample.seednumber=7;
hive> SELECT *
    > FROM numbers
    > TABLESAMPLE(20.0 PERCENT);
OK
numbers.n
7
8
Time taken: 0.031 seconds, Fetched: 2 row(s)

まとめ

  • Apache Hive でサンプリングするときは TABLESAMPLE を使う
  • サンプリングの挙動について
    • 特定の値をハッシュ化してバケットに振り分ける
    • 振り分けられたバケットを選択する
    • ハッシュ化に使う値は rand() 関数の値や、特定のカラムの内容が使える
  • 上記のやり方の他にブロックサンプリングという方法もある
    • ブロック単位でサンプリングされる点に注意が必要となる
    • サンプリング結果を変えたいときは hive.sample.seednumber を変更する

プログラミング Hive

プログラミング Hive

  • 作者: Edward Capriolo,Dean Wampler,Jason Rutherglen,佐藤直生,嶋内翔,Sky株式会社玉川竜司
  • 出版社/メーカー: オライリージャパン
  • 発売日: 2013/06/15
  • メディア: 大型本
  • この商品を含むブログ (3件) を見る