The pursuit of happiness with Python and PostgreSQL®の翻訳です。
2021年6月23日
PythonとPostgreSQL®による幸福の追求
世界とその中のデータが完璧であることはめったにありません。PostgreSQL®データベースのデータを分析に使えるようにクリーンアップする方法をご覧ください。
情報の作成は、創造的なプロセスと同じです。材料（この場合はデータソース）を見つけることから始まり、データセットをクリーニングし、結合し、整理します。完成する頃には、美しいものを作り上げ、共有すべき新たな洞察を得ている。
情報作成において最も広く使われている言語のひとつがPythonであり、データサイエンティスト、エンジニア、アナリストに愛されている。
このブログでは、そのようなライブラリを3つ紹介する：pandas、ddlgenerator、psycopg2`である。これらはそれぞれ、データセットをクリーンにしてPostgreSQLデータベースにプッシュし、後でそのデータをクエリして様々な企業の数字を表示できるようにするものだ。
今日の目的は単純で、幸福のデータセットを探索し、楽しい人生を送るためには世界のどこに移動すべきかを見つけることである！
大まかな流れはこうだ：
1.データセットを取得する
2.データのクリーニング
3.PostgreSQLインスタンスの作成
4.データをPostgreSQLにプッシュする
5.PostgreSQLにデータを問い合わせる
0. データセットを取得する
前述したように、データセットを見つけることから始める。Kaggleはデータサイエンスのコミュニティで広く使われているウェブサイトで、チャレンジやコンペ、学習に使われるデータセットを提供している。ログインするだけでダウンロードできるWorld Happinessに関する素晴らしいデータセットがある。このデータセットには、5つのCSVファイルが含まれており、1年ごとに1つずつ、様々な国の幸福度ランキングとその他の指標が掲載されている。
happinessという名前のフォルダと、ファイルを格納するためのdata`という名前のサブフォルダを作成しよう：
mkdir -p happiness/data
cd happiness
次に、Kaggleからデータをダウンロードし、5つのCSVをhappiness/dataフォルダに置きます。
1.データのクリーンアップ
よくあることだが、ソース・データセットが常に整頓されたきれいなものであるとは限らない。これは我々の幸福のデータセットにも当てはまる。ファイルをブラウズすると、2015.csv と 2016.csv はカラム名とデータのフォーマットが同じであることがわかる。しかし、2017.csvにはRegionカラムがなく、他のカラムの名前と順序が異なっている（GenerosityとTrustが入れ替わっている）。同じことが2018.csvと2019.csvのファイルにも当てはまり、信頼区間を見つけることができなくなっている。さらに、年はファイル名にのみ含まれており、列には含まれていません。
クリーニングを始める前に、新しいPythonプロジェクトを始めるたびに別の仮想環境を作成するのは良い習慣です。こうすることで、デフォルトのPython環境が競合するライブラリで過負荷にならないようにします。仮想環境を作成し、次のように有効にします：
python -m venv python_pg
source python_pg/bin/activate
データの一貫したビューを生成するために、ファイルにはいくつかの作業が必要である。ターミナルで以下のコードを実行して、データ操作と分析に役立つpandasライブラリをインストールすることから始めよう：
pip install pandas
これで happiness フォルダに prepare.py というファイルを作成し、その中に Python のコードを含めることができます。デフォルトでインストールされているはずの os と glob と共に、先ほどインストールしたライブラリをインポートすることから始めましょう。
import pandas as pd
import glob, os
次に、一度クリーニングしたデータセットを格納するために、all_years_df という名前の DataFrame で使用するカラム名を定義します。次の行を prepare.py ファイルに追加します：
column_names = ['survey_yr', 'country', 'overall_rank'、
'score', 'gdp', 'generosity', 'freedom'、
social_support', 'life_exp', 'gov_trust'］
all_years_df = pd.DataFrame()`
さて、次は混沌に秩序をもたらし、dataフォルダ内のすべてのCSVファイルを解析する番だ。次のコードを prepare.py ファイルに追加します：
files = glob.glob('data/*.csv')
for fp in files：
# ファイル名を新しいカラムとして扱う
file_year = os.path.basename(fp).split('.')[0］
file_content = pd.read_csv(fp).assign(Year=int(file_year)-2000)
# 各ファイルについて、年に基づいて正しい列を選ぶ
uniformed_columns = find_columns(file_content, file_year)
# カラム名を適合し、解析された年のデータをall_years
uniformed_columns.columns = column_names
all_years_df = pd.concat([all_years_df, uniformed_columns])
all_years_df.to_csv('results.csv', index=False)
glob関数は、happiness/dataフォルダにあるファイルのリストを返す。各ファイルについて、os.path.basenameでファイル名を読み取り、ドット(.)で分割して最初の部分を取り出し、年の情報のみを抽出する。file_content 行では、pandas の read_csv 関数を使用してファイルデータを読み込み、新しいカラム Year に（assign 関数を使用して）西暦の下 2 桁（int(file_year)-2000）を代入する。
次に、すぐに定義する find_columns 関数を呼び出して、ソースファイルから年によって適切なカラムを選択する。
その後、1年分のデータのみを含む uniformed_columns DataFrame にカラム名を統合し、最後にデータセット全体を含む all_years_df DataFrame に追加します。最後に、最終結果を happiness フォルダ内の results.csv という名前の CSV ファイルに格納する。
正しいカラムの選択
足りない部分は find_columns 関数の定義で、ここでカラム名と順序を統合します。上述したように、年によって、ソースファイルから正しいカラムを選択する必要があります。以下のコードを prepare.py ファイルの import ステートメントの直後に追加して、様々な年に対してトリックを実行します：
def find_columns(file_content, file_year):
if file_year in ['2015', '2016']:
uniformed_columns = file_content[[
'Year', 'Country', 'Happiness Rank', 'Happiness Score',
'Economy (GDP per Capita)', 'Generosity', 'Freedom', 'Family',
'Health (Life Expectancy)', 'Trust (Government Corruption)'
]]
elif file_year in ['2017']:
uniformed_columns = file_content[[
'Year', 'Country', 'Happiness.Rank', 'Happiness.Score',
'Economy..GDP.per.Capita.', 'Generosity', 'Freedom', 'Family',
'Health..Life.Expectancy.', 'Trust..Government.Corruption.'
]]
else:
uniformed_columns = file_content[[
'Year', 'Country or region', 'Overall rank', 'Score',
'GDP per capita', 'Generosity', 'Freedom to make life choices',
'Social support', 'Healthy life expectancy',
'Perceptions of corruption'
]]
return uniformed_columns
この関数は、各年の正しいカラムを選択し、結果のDataFrameを返します。
パイプラインが機能していることを確認しましょう。ターミナルを happiness フォルダに置き、次のように実行します：
python prepare.py
次に happiness フォルダにある results.csv ファイルをチェックして、コードが動作したことを確認します。結果はこのようになっているはずです：
2.PostgreSQLインスタンスの作成
2つ目のステップは、データをPostgreSQLのテーブルに格納することです。データを保存する前に、PGデータベースが必要です。ターミナルでAiven CLIを使って作成しましょう。
avn service create demo-pg \
-t pg \
--cloud google-europe-west3 \
-p hobbyist
上記のコマンドは hobbyist プランを使用して google-europe-west3 上に demo-pg という名前の PostgreSQL インスタンスを作成します。全てのPostgreSQLプランについては価格ページで確認できる。接続するには、host, port, デフォルトの avnadmin ユーザの password のような情報を含む service uri が必要である。これを取得するには
avn service get demo-pg --format '{service_uri}'
ターミナルで次のコマンドを実行し、インスタンスの準備が整うのを待つ間、数分間くつろいでいよう。
avn service wait demo-pg
3.PostgreSQL にデータをプッシュする
サービスが実行されたら、Pythonに戻ってデータをプッシュする準備をする。
PostgreSQLのテーブルを作成するSQL文を自動生成するためにddlgeneratorライブラリを使い、データベースとやりとりするためにpsycopg2を使います。ターミナルで以下の行を実行することで、両方をインストールできる：
pip install psycopg2 ddlgenerator
新しい push.py ファイルを作成し、import を追加してデータベースに接続します。以下のコードの <SERVICE_URI_OUTPUT> パラメータを上記の avn service get コマンドの出力に置き換えてください：
import psycopg2
conn = psycopg2.connect('<SERVICE_URI_OUTPUT>')
接続が確立されました。これでテーブルを作成し、データをプッシュできるようになりました。ddlgeneratorは、CSVファイルからテーブルDDLとinsertステートメントの両方を作成する素晴らしい機能を提供する。以下のコードを push.py` ファイルに追加します：
from ddlgenerator.ddlgenerator import Table
table = Table('results.csv', table_name='HAPPINESS')
sql = table.sql('postgresql', inserts=True)
ddlgeneratorからTable クラスをインポートした後、results.csvファイルを解析するインスタンスを作成する。最後の行では、PostgreSQL用のCREATE` テーブル定義SQLを作成している。SQL文を表示すると以下のようになる：
DROP TABLE IF EXISTS happiness；
CREATE TABLE happiness (
survey_yr INTEGER NOT NULL、
country VARCHAR(24) NOT NULL、
overall_rank INTEGER NOT NULL、
score DECIMAL(15, 14) NOT NULL、
gdp DECIMAL(17, 16) NOT NULL、
寛大さ DECIMAL(16, 16) NOT NULL、
自由度 DECIMAL(16, 16) NOT NULL、
social_support DECIMAL(16, 15) NOT NULL、
life_exp DECIMAL(17, 16) NOT NULL、
gov_trust DECIMAL(16, 16)
);
INSERT INTO happiness (survey_yr, country, overall_rank, score, gdp, generosity, freedom, social_support, life_exp, gov_trust) VALUES (19, 'Finland', 1, 7.769, 1.34, 0.153, 0.596, 1.587, 0.986, 0.393)；
INSERT INTO happiness (survey_yr, country, overall_rank, score, gdp, generosity, freedom, social_support, life_exp, gov_trust) VALUES (19, 'Denmark', 2, 7.6, 1.383, 0.252, 0.592, 1.573, 0.996, 0.41)；
次のコードを push.py ファイルに追加することで、データベースに対して sql 文を実行することができます。
cur = conn.cursor()
cur.execute(sql)
conn.commit()
cur.close()
conn.close()
カーソルを作成し、SQL文を実行し、変更をデータベースにコミットし、カーソルと接続を閉じました。いよいよコードを実行します。同じターミナルでhappinessフォルダ上で以下を実行してください：
python push.py
これで PostgreSQL にデータがロードされました。
4.PostgreSQLでデータを照会する
PostgreSQL の happiness テーブルにデータが正しく格納されているか確認してみましょう。新しいPythonファイルquery.pyを以下の内容で作成します(先ほどと同様に、接続のservice uriを編集します)：
import psycopg2
conn = psycopg2.connect('<SERVICE_URI_OUTPUT>')
cur = conn.cursor()
cur.execute('')
select survey_yr, country, overall_rank
from HAPPINESS
where overall_rank <= 3
order by survey_yr, overall_rank
''')
print(cur.fetchall())
cur.close()
conn.close()
push.py`ファイルにあるように、PostgreSQLへの接続を作成し、クエリを実行して様々な年の上位3位までの国をチェックしています。結果は以下のようになります：
[
(15, 'Switzerland', 1), (15, 'Iceland', 2), (15, 'Denmark', 3),
(16, 'Denmark', 1), (16, 'Switzerland', 2), (16, 'Iceland', 3),
(17, 'Norway', 1), (17, 'Denmark', 2), (17, 'Iceland', 3),
(18, 'Finland', 1), (18, 'Norway', 2), (18, 'Denmark', 3),
(19, 'Finland', 1), (19, 'Denmark', 2), (19, 'Norway', 3)
]
北欧諸国は住みやすそうだ！特に、Aivenの母国であるフィンランドが、過去2年間で首位に立っていることに注目してほしい。
このデータは、PostgreSQLのリレーショナル・テーブルで利用できるようになり、主要な分析ツールやレポート作成ツールで照会して公開することができるため、より多くの人々がアクセスできるようになった。
まとめ
適切なデータセットを見つけることは始まりに過ぎない。データをクリーニングし、整理し、公開することで、人々の意思決定に役立つ情報が生まれる。Pythonはこの分野に非常に適しており、データコミュニティで広く採用されている。ブログで紹介した pandas、ddlgenerator、psycopg2 ライブラリを使えば、簡単にデータを扱い、PostgreSQL にプッシュすることができる。これはデータディスカバリーや企業レポーティングのための強固な基盤になる。
もう少し詳しい情報を：
- Aiven for PostgreSQL: Aivenが提供するすべての情報
- Aiven PostgreSQL supported extensions: Aivenがサポートする拡張機能をご覧いただけます。
- pandas、ddlgenerator、psycopg2: 使用ライブラリに関するドキュメント
--
まだAivenのサービスをご利用になっていませんか？https://console.aiven.io/signupから無料トライアルにお申し込みください！
また、changelogやblogのRSSフィード、またはLinkedInやTwitterのアカウントをフォローし、製品や機能関連の最新情報をご確認ください。