以前TianChiの紹介編と実践編をご紹介させて頂きましたが、今回TianChiシリーズの最終編（データ処理）に関して、まとめていきたいと思います。
データセットの前処理と言えば、オープンソースのPandasでDataframeとSeriesを操作するのが一般的ですが（TianChiシリーズの実践編でご参照ください）、データのサイズが大きいほど、大量のデータを読み込もうとすると、メモリ不足でエラーになる可能性、もしくは長時間待たされる可能性も高くなります。
Tianchiに開催されるコンペティションは実に様々で、提供されているデータセットが、かならずしもビッグデータの規模とは言えないですが、本記事では、ビッグデータの前処理に焦点をあてながら、PyODPSで2次元テーブル、そしてテーブルの列からデータの取り扱い方について、展開させて頂きたいと思います。
PyODPSとは？
PyODPS はAlibaba Cloudによって独自に開発されたビッグデータ処理プラットフォームであるMaxComputeのPython SDK です。また、MaxCompute でのビッグデータの前処理と分析を容易にするために、独自のDataFrameフレームワークも提供しています。
Pandasライクなインターフェースや、 DataFrame API をサポートしつつ、 大量のデータをMaxComputeで並列分散処理の機能を活用することができるのが、PyODPSの一番大きなメリットだと言えます。Pandasに既に慣れ親しんだ方にとっては、非常に手をつけやすいツールだと思います。
MaxComputeについて
2010年から、Alibaba Cloudが自社の日々増え続ける膨大なデータを格納・処理するために、独自に研究開発した分散コンピューティング処理ベースのデータウェアハウスソリューションです。2014年に自社のビッグデータ系の主力なクラウドサービスとして公開して以来、日本でもお馴染みになった中国11月11日の「独身の日」セールでのビッグデータ処理などに活用されています。
また、従来分散コンピューティングでは、データ分析には専門的な IT 知識が必要になりますが、MaxCompute が提供するデータ連携ソリューションにより、ユーザーは分散コンピューティングの詳細を意識することなくデータ分析を行うことも可能です。
データのアップロード
MaxComputeへのデータのアップロードに対して、データソースによって、実は複数の方法がありますが、今回はMaxComputeのクライエントであるtunnelを使い、データのアップロード方法を説明します。
具体的に、Alibaba CloudがTianchiで投稿したコンペ『Offline to Online (O2O) Prediction of Coupon Redemption』を例とし、提供されているデータセット(ccf_offline_stage1_train.zip)をまずローカルにダウンロードします。zipファイルを展開すると、ccf_offline_stage1_trainという名前のcsvファイルがあります。
まず、MaxComputeで、データセットに対応するODPSテーブルを作成します。
次は、以下のtunnel コマンドでデータセットを先ほど述べたODPSテーブルにアップロードします。
tunnel upload ~/Downloads/dataset/ccf_offline_stage1_train.csv yourProject.ccf_offline_stage1_train;データの前処理
まず、Alibaba Cloud アカウント（AccessKey IdとAccessKey Secretがユーザー情報管理でご確認ください。）を使用し、 MaxCompute エントリを初期化します。初期化が完了すると、テーブルを操作できるようになります。
>>>from odps import ODPS
>>>odps = ODPS('accesskey_id', 'accesskey_secret',
'project_name', endpoint = 'endpoint')下記のコマンドを呼び出し、テーブルの情報を確認することができます。
>>>t = odps.get_table('ccf_offline_stage1_train')
>>>print(t.schema)
odps.Schema {
user_id string
merchant_id string
coupon_id string
discount_rate string
distance string
date_received string
date_used string
}DataFrame オブジェクトを作成するには、DataFrameモジュールを入れてから、下記のコマンドを実行します。
>>>from odps.df import DataFrame
>>>dfoff = DataFrame(odps.get_table('ccf_offline_stage1_train'))Pandasライクなhead メソッドを使用すると、最初の N 個のデータレコードを取得し、素早くデータをプレビューすることができます。
>>>dfoff.head(10)| User_id | Merchant_id | Coupon_id | Discount_rate | Distance | Date_received | Date_used | |
| 0 | 1439408 | 2632 | null | null | 0 | null | 20160217 |
| 1 | 1439408 | 4663 | 11002 | 150:20:00 | 1 | 20160528 | null |
| 2 | 1439408 | 2632 | 8591 | 20:01 | 0 | 20160217 | null |
| 3 | 1439408 | 2632 | 1078 | 20:01 | 0 | 20160319 | null |
| 4 | 1439408 | 2632 | 8591 | 20:01 | 0 | 20160613 | null |
| 5 | 1439408 | 2632 | null | null | 0 | null | 20160516 |
| 6 | 1439408 | 2632 | 8591 | 20:01 | 0 | 20160516 | 20160613 |
| 7 | 1832624 | 3381 | 7610 | 200:20:00 | 0 | 20160429 | null |
| 8 | 2029232 | 3381 | 11951 | 200:20:00 | 1 | 20160129 | null |
| 9 | 2029232 | 450 | 1532 | 30:05:00 | 0 | 20160530 | null |
Pandasの基本的なデータ構造（DataFrame と Series）と同様に、PyODPSのDataFrameフレームワークにも、Collectionと呼ばれる二次元のテーブルとSequenceと呼ばれる一次元の配列があります。ただここに気をつけて欲しいのは、Seriesは二次元テーブルの列(column)か行(row)を指すのに対して、Sequenceは二次元テーブルの列のみと指しますので、行をSequenceとしてデータを取り出すことができません。
| Operation | Pandas | PyODPS |
| Select column | df[col] | df[col] , df.col |
| Select row by label | df.loc[label] | × |
| Select row by integer location | df.iloc[loc] | × |
PyOPDS Dataframに１つの便利な機能としては、to_pandas()関数を活用することによって、PyODPSで処理した結果をPandas Dataframeの形式で保存することができます。また、MatplotlibやSeabornなどのライブラリを一緒に使うと、描画の機能も実現することができます。
import matplotlib.pyplot as plt
import seaborn as sns
sns.set_style('darkgrid')
sns.set_context("notebook", font_scale= 1.4)
x = dfoff.distance.unique().to_pandas().values
y = dfoff.groupby('distance').agg(count=dfoff.distance.count()).to_pandas()
plt.yscale('log')
plt.ylabel('Distance Count')
plt.bar(x,y['count'].values)
plt.tight_layout()
参考サイト：
本記事に関連するMaxCompute、PyODPSのオープンソースやTunnelに関しての詳しい設定方法は以下リンクをご参照ください。