いつもご覧いただき誠にありがとうございます。
ご存知のとおり？弊社は世界最大級のイノベーションデータベースを保有しており、中にはクラウドファンディング、科研費データ、特許データなど様々なデータが含まれてます。
普段仕事上データを入手してから弊社DBに入れるまでのプロセス（所謂Data Preparation的な？）を担当するので、今日はその辺で役たちそうなネタにしようかと思います。
データの種類が多い分、それぞれ元データの入手ルートによってはノイズが含まれていることもございます。
例えば科研データはクォリティが高いのですが、クラウドファンディングはデータの性質（個人もプロジェクトを起こすことが出来る）上どうしてもノイズが含まれたりする場合があります。
そのようなデータを弊社データベースに入れる前にデータの品質管理をきちんと行い、ノイズを排除する必要があります。
その為のツールとしてデータクリーニング系のオープンソースを探してますが、現状オープンソースの製品がそれほど多くないのが実情であって私自身もどれを活用すればいいのか結構迷ってます。
そこで今日はデータクォリティをチェックできるオープンソースのツール：Optimusを例を交えながら紹介したいと思います。
Optimus
Spark(PySpark)と連携してデータのクォリティをチェック、データクレンジング、変換ができるツールです。
正式リリースして間もないので、機能面でまだこれからという感じではありますが、
この類のopen sourceがまだまだ少ないので今回試してみることになりました。
Github : https://github.com/ironmussa/Optimus
事前準備
環境
- Apache Spark 2.2
- Python >= 3.5
pipからインストール
pip install optimuspyspark
- Spark packageとして利用する
※まず、python 3.5以上入れる必要があります。
export PYSPARK_PYTHON=python3 // Pysparkのpythonバージョンを3.5以上に設定する pyspark2 --packages ironmussa:Optimus:1.0.3
- 作業の便宜上jupyterも入れて、起動する
pip3 install jupyter jupyter notebook --allow-root --ip='*'
動かしてみる
基本的にサンプルはオフィシャルサイトに載ってありますが、そのままだと動かなかったので（たぶんドキュメントが古い？）
ちょっとhelpで仕様探ったりしてました。
今回はクラウドファンディングのデータを用意して色々試してみたいと思います。
以下データ一部
+----------+--------------------+-----------+--------------------+-----------+-------------+-------------+ |project_id| name| category| location|goal_amount|raised_amount|backers_count| +----------+--------------------+-----------+--------------------+-----------+-------------+-------------+ | 4876| Spirit of the Game| Film|London, Stoke, Bl...| 5000| 335| 5| | 1006067| FoodQuest| Food|Wiener Neustadt, ...| 25000| 25| 2| | 814| Connected| Film|San Francisco, Ca...| 10000| 112| 4| | 53500| Arcadian| Film|Orlando, Florida,...| 1500| 400| 5| | 339731|Help me build out...|Photography|Austell, Georgia,...| 5000| 115| 4| | 380670|The Improvisation...| Music|Portland, Oregon,...| 2000| 1689| 35| | 1932270| The Lonely Angel| Film|San Jose, Califor...| 5000| 2115| 27| | 1300172| The House Lady| Community|Knoxville, Tennes...| 15000| 58| 7| | 42509| Mindless Series|Video / Web|Central Coast NSW...| 500| 130| 3| | 814777| Spencer Zahrn Art| Art|Little Rock, Arka...| 1200| 1400| 16| +----------+--------------------+-----------+--------------------+-----------+-------------+-------------+
ではまず上記のデータの統計情報(stats的な)を取ってみたいと思います。
jupyter起動して以下のコマンドを打ちます
import optimus as op import os
console上以下のメッセージが出るまで待ちます
Starting or getting SparkSession and SparkContext. Setting checkpoint folder (local). If you are in a cluster change it with set_check_point_folder(path,'hadoop'). Deleting previous folder if exists... Creation of checkpoint directory... Done.
テストデータを読み込んで全カラムの分析を行う
tools = op.Utilities()
filePath = "file:///" + os.getcwd() + "/test/cf_sample.csv"
df = tools.read_dataset_csv(path=filePath,delimiter_mark=',')
analyzer = op.DataFrameAnalyzer(df=df,path_file=filePath)
analyzer.column_analyze("*", plots=False, values_bar=True, print_type=False, num_bars=10) // 全カラムの分析を実施
全カラムの分析を実施すると下記結果のように各カラムごとデータ型毎の件数、Min valueとMax valueなどが出力されることが分かります。
※'*'の代わりに分析したいカラム名だけ書けばそのカラムだけ分析できます。
結果一部
上記の結果からgoal_amountのカラムにはintegerとstringタイプが混在していて、nullの値も含まれていることが分かります。
続いてカラムごとの詳細分析：値の出現頻度を調べます。
categoryDf = analyzer.get_data_frame().select("category") // 詳細分析したいカラム
hist_dictCategory = analyzer.get_categorical_hist(df_one_col=categoryDf, num_bars=10) //文字列ごとの出現頻度
print(hist_dictCategory)
結果
[{'value': 'Film', 'cont': 215}, {'value': 'Community', 'cont': 127}, {'value': 'Music', 'cont': 102}, {'value': 'Education', 'cont': 72}, {'value': 'Health', 'cont': 55}, {'value': 'Small Business', 'cont': 49}, {'value': 'Theatre', 'cont': 48}, {'value': 'Technology', 'cont': 42}, {'value': 'Video / Web', 'cont': 36}, {'value': 'Art', 'cont': 33}]
以下のコードでグラフ化もできます
analyzer.plot_hist(df_one_col=categoryDf,hist_dict= hist_dictCategory, type_hist='categorical')
カラムごとのユニーク値をカウント
print(analyzer.unique_values_col("location"))
結果
{'total': 1000, 'unique': 555}
上記以外にも色んな機能が使えますので、詳細はhelp(analyzer)でご確認ください
データ変換
データ変換はDataFrameTransformerクラスで提供してます
主には以下の機能が提供されてます
対カラム(column)：
- カラムのdrop
- replace
- rename
- move（ポジション変更）
対行(Row)：
- 行のdrop
- delete
個別のカラムに対して：
- trim
- accentのクリア
- lookup
- 特殊文字の削除
- 日付のフォーマット変換
詳細はGithubにありますので、 ここでは一部分だけピックアップして説明したいと思います
transformer = op.DataFrameTransformer(df) transformer.show() +----------+--------------------+-----------+--------------------+-----------+-------------+-------------+ |project_id| name| category| location|goal_amount|raised_amount|backers_count| +----------+--------------------+-----------+--------------------+-----------+-------------+-------------+ | 4876| Spirit of the Game| Film|London, Stoke, Bl...| 5000| 335| 5| | 1006067| FoodQuest| Food|Wiener Neustadt, ...| 25000| 25| 2| | 814| Connected| Film|San Francisco, Ca...| 10000| 112| 4| | 53500| Arcadian| Film|Orlando, Florida,...| 1500| 400| 5| | 339731|Help me build out...|Photography|Austell, Georgia,...| 5000| 115| 4| | 380670|The Improvisation...| Music|Portland, Oregon,...| 2000| 1689| 35| | 1932270| The Lonely Angel| Film|San Jose, Califor...| 5000| 2115| 27| | 1300172| The House Lady| Community|Knoxville, Tennes...| 15000| 58| 7| | 42509| Mindless Series|Video / Web|Central Coast NSW...| 500| 130| 3| | 814777| Spencer Zahrn Art| Art|Little Rock, Arka...| 1200| 1400| 16| +----------+--------------------+-----------+--------------------+-----------+-------------+-------------+
以下の例ではreplace_colを使ってcategoryのカラムからFilmの文字列をMovieに置換します。
transformer.replace_col(search='Film', change_to='Movie', columns='category') <optimus.df_transformer.DataFrameTransformer at 0x7f7243ae0cc0> // エラーっぽいのが出てるけど処理自体は想定とおり動いたので、とりあえず気にしません transformer.show() // 変更後 +----------+--------------------+-----------+--------------------+-----------+-------------+-------------+ |project_id| name| category| location|goal_amount|raised_amount|backers_count| +----------+--------------------+-----------+--------------------+-----------+-------------+-------------+ | 4876| Spirit of the Game| Movie|London, Stoke, Bl...| 5000| 335| 5| | 1006067| FoodQuest| Food|Wiener Neustadt, ...| 25000| 25| 2| | 814| Connected| Movie|San Francisco, Ca...| 10000| 112| 4| | 53500| Arcadian| Movie|Orlando, Florida,...| 1500| 400| 5| | 339731|Help me build out...|Photography|Austell, Georgia,...| 5000| 115| 4| | 380670|The Improvisation...| Music|Portland, Oregon,...| 2000| 1689| 35| | 1932270| The Lonely Angel| Movie|San Jose, Califor...| 5000| 2115| 27| | 1300172| The House Lady| Community|Knoxville, Tennes...| 15000| 58| 7| | 42509| Mindless Series|Video / Web|Central Coast NSW...| 500| 130| 3| | 814777| Spencer Zahrn Art| Art|Little Rock, Arka...| 1200| 1400| 16| +----------+--------------------+-----------+--------------------+-----------+-------------+-------------+
以下の例ではTransformer.delete_row(func)をもちいて条件に合致するレコードを削除する
from pyspark.sql.functions import col
func = lambda g: (g > 5000) & (g <= 200000) //残したい条件を設定する
transformer.delete_row(func(col('goal_amount')))
ここではgoal_amountが5000より大きく、200000以下のレコードを残すようにしたので、結果はこんな感じになります
+----------+--------------------+--------------+--------------------+-----------+-------------+-------------+ |project_id| name| category| location|goal_amount|raised_amount|backers_count| +----------+--------------------+--------------+--------------------+-----------+-------------+-------------+ | 1006067| FoodQuest| Food|Wiener Neustadt, ...| 25000| 25| 2| | 814| Connected| Movie|San Francisco, Ca...| 10000| 112| 4| | 1300172| The House Lady| Community|Knoxville, Tennes...| 15000| 58| 7| | 783762|A Story Unwritten...| Movie| Barcelona, Spain| 7500| 7925| 127| | 781986|Die deutsch - chi...| Education| Berlin, Germany| 12000| 450| 5| | 475016|6:38: THE DEATH O...| Movie|Los Angeles, Cali...| 200000| 390| 7| | 370437| SLOW WIN| Movie|Toronto, Ontario,...| 10000| 740| 19| | 53835|The Finger and th...| Art| Genoa, Italy| 15000| 150| 4| | 513792|Intrusion Disconn...| Movie|Louisville, Kentu...| 10000| 195| 4| | 1188584|Wild West Steel -...|Small Business|Idaho Falls, Idah...| 50000| 30| 2| +----------+--------------------+--------------+--------------------+-----------+-------------+-------------+
Transformer.set_col(columns, func, dataType)を使ってセルの変換を行う
func = lambda backers_count: 10 if (backers_count < 10 ) else backers_count transformer.set_col(['backers_count'], func, 'integer')
10より小さいbackers_countの値を10に変換するので、以下のように変換されます
+----------+--------------------+--------------+--------------------+-----------+-------------+-------------+ |project_id| name| category| location|goal_amount|raised_amount|backers_count| +----------+--------------------+--------------+--------------------+-----------+-------------+-------------+ | 1006067| FoodQuest| Food|Wiener Neustadt, ...| 25000| 25| 10| | 814| Connected| Movie|San Francisco, Ca...| 10000| 112| 10| | 1300172| The House Lady| Community|Knoxville, Tennes...| 15000| 58| 10| | 783762|A Story Unwritten...| Movie| Barcelona, Spain| 7500| 7925| 127| | 781986|Die deutsch - chi...| Education| Berlin, Germany| 12000| 450| 10| | 475016|6:38: THE DEATH O...| Movie|Los Angeles, Cali...| 200000| 390| 10| | 370437| SLOW WIN| Movie|Toronto, Ontario,...| 10000| 740| 19| | 53835|The Finger and th...| Art| Genoa, Italy| 15000| 150| 10| | 513792|Intrusion Disconn...| Movie|Louisville, Kentu...| 10000| 195| 10| | 1188584|Wild West Steel -...|Small Business|Idaho Falls, Idah...| 50000| 30| 10| +----------+--------------------+--------------+--------------------+-----------+-------------+-------------+
次はcategoryをすべて大文字に変換することを試します
func = lambda category: category.upper() transformer.set_col(['category'], func, 'string')
結果
+----------+--------------------+--------------+--------------------+-----------+-------------+-------------+ |project_id| name| category| location|goal_amount|raised_amount|backers_count| +----------+--------------------+--------------+--------------------+-----------+-------------+-------------+ | 1006067| FoodQuest| FOOD|Wiener Neustadt, ...| 25000| 25| 10| | 814| Connected| MOVIE|San Francisco, Ca...| 10000| 112| 10| | 1300172| The House Lady| COMMUNITY|Knoxville, Tennes...| 15000| 58| 10| | 783762|A Story Unwritten...| MOVIE| Barcelona, Spain| 7500| 7925| 127| | 781986|Die deutsch - chi...| EDUCATION| Berlin, Germany| 12000| 450| 10| | 475016|6:38: THE DEATH O...| MOVIE|Los Angeles, Cali...| 200000| 390| 10| | 370437| SLOW WIN| MOVIE|Toronto, Ontario,...| 10000| 740| 19| | 53835|The Finger and th...| ART| Genoa, Italy| 15000| 150| 10| | 513792|Intrusion Disconn...| MOVIE|Louisville, Kentu...| 10000| 195| 10| | 1188584|Wild West Steel -...|SMALL BUSINESS|Idaho Falls, Idah...| 50000| 30| 10| +----------+--------------------+--------------+--------------------+-----------+-------------+-------------+
ダラダラと長く書いたのですが、以下のように短いコードで簡単なデータ変換が出来ます
transformer.trim_col("*")
.remove_special_chars("*")
.clear_accents("*")
.lower_case("*")
.drop_col("dummyCol")
.set_col(func)
.delete_row(func)
OutlierDetectorクラスを使って外れ値のチェック
以下例
from pyspark.sql.types import StringType, IntegerType, StructType, StructField
schema = StructType([
StructField("name", StringType(), True),
StructField("category", StringType(), True),
StructField("status", IntegerType(), True)])
name = ['Spirit of the Game', 'FoodQuest', 'Connected', 'Arcadian']
category = ['Food', 'Movie', 'Photography', 'Movie']
status = [1,1,1,9]
df = op.spark.createDataFrame(list(zip(name, category, status)), schema=schema)
transformer = op.DataFrameTransformer(df)
transformer.show()
+------------------+-----------+------+
| name| category|status|
+------------------+-----------+------+
|Spirit of the Game| Food| 1|
| FoodQuest| Movie| 1|
| Connected|Photography| 1|
| Arcadian| Movie| 9|
+------------------+-----------+------+
detector = op.OutlierDetector(df,"num")
detector.outliers()
[9] <- 外れ値
まとめ
OptimusはSpark上手軽にデータ品質分析、変換が可能なオープンソースで、
この種類のツールが少ないので、非常に貴重だなと感じました。
サンプル類もたくさん用意されているので入門のハードルも低く感じました。
まだ正式リリースしたばかりなので、動かしてみるとドキュメントとおり行かない部分とかもありましたが、 helpで仕様確認しながらやれば特に困ることもなかったので、これからが楽しみなツールです。
ではまた！
