pandasで外部データベースのデータを取得する

この記事は GMOアドマーケティング Advent Calendar 2019 10日目の記事です。

はじめに

GMOアドマーケティングのS.Sです。
簡単なデータの集計などのためにpandasを使う場面はあるかと思いますが、外部のデータベースの集計結果をcsvファイルに保存してjupyter notebookなどで読み込んでさらに集計や可視化などをするケースだとやや回りくどい感じがします。
ほかにもデータがMySQLやBigQueryなど複数箇所に分散しているケースで両方の結果を使って集計したい場面でかゆいところに手が届かないことがあります。

pandasやそのほかのライブラリなどを用いるとMySQLやBigQuery上のテーブルに対する集計結果をそのままDataFrameとして読み込める機能があるので今回の記事ではその機能について紹介したいと思います。

外部データベースからのデータ取得(MySQL)

はじめに必要なライブラリをインストールします。

pip install pandas
pip install sqlalchemy
pip install mysql-connector
pip install --upgrade google-cloud-bigquery

pip install pandas

pip install sqlalchemy

pip install mysql-connector

pip install --upgrade google-cloud-bigquery

まずはSQLAlchemyを利用してMySQLのデータベースから集計結果を取得するクエリを書いてみます。

from sqlalchemy import create_engine
import pandas as pd
import getpass
uri = "mysql+mysqlconnector://user:{}@host:port/db?charset=utf8".format(getpass.getpass())
engine = create_engine(uri)
df = pd.read_sql_query("SELECT group_id, COUNT(*) AS count FROM table GROUP BY group_id", engine)

from sqlalchemy import create_engine

import pandas as pd

import getpass

uri = "mysql+mysqlconnector://user:{}@host:port/db?charset=utf8".format(getpass.getpass())

engine = create_engine(uri)

df = pd.read_sql_query("SELECT group_id, COUNT(*) AS count FROM table GROUP BY group_id", engine)

パスワードはノートブックに直書きすると共有するときなどに消し忘れて困るので、getpassを使います。そうするとjupyter notebookのセルを実行したときにpromptが表示されてパスワードを入力できます。
user, host, ポート番号などは適宜置き換えてください。
SQLAlchemyのengineインスタンスをpandasのread_sql_query関数に渡すと、MySQLでクエリを実行した集計結果がpandas DataFrameに格納されて取得できます。
group_idをカラムとして持つようなテーブルに対して上記のクエリを実行すると次のような結果が表示されます。

次のようにSSHでトンネルをつくると踏み台を経由してサーバーに接続することもできます。

ssh -N -L local_port:remote_host:remote_port user@host

1	ssh -N -L local_port:remote_host:remote_port user@host

外部データベースからのデータ取得(BigQuery)

BigQueryに対してクエリを投げてpandasのDataFrameとして取得する方法もみていきます。

from google.cloud import bigquery
client = bigquery.Client()

query = "SELECT group_id, COUNT(*) AS count FROM `dataset.table` GROUP BY group_id"
query_job = client.query(query)
res = query_job.result()
df = res.to_dataframe()

from google.cloud import bigquery

client = bigquery.Client()

query = "SELECT group_id, COUNT(*) AS count FROM `dataset.table` GROUP BY group_id"

query_job = client.query(query)

res = query_job.result()

df = res.to_dataframe()

clientオブジェクトをつくり、queryメソッドにクエリを渡すとBigQueryでの集計結果が取得できます。
結果のオブジェクトはto_dataframeを呼ぶとpandasのDataFrameに変換できます。

BigQueryでクエリを投げる場合は料金も気になりますが、クエリのデータ処理量は次のような形で確認できます。

def bytes_processed(cl, q):
    job_config = bigquery.QueryJobConfig()
    job_config.dry_run = True
    job_config.use_query_cache = False
    s = cl.query(q, job_config=job_config)
    s.dry_run
    return s.total_bytes_processed / (1024 ** 3), "GB"

bytes_processed(client, query)

def bytes_processed(cl, q):

job_config = bigquery.QueryJobConfig()

job_config.dry_run = True

job_config.use_query_cache = False

s = cl.query(q, job_config=job_config)

s.dry_run

return s.total_bytes_processed / (1024 ** 3), "GB"

bytes_processed(client, query)

今回の記事ではMySQLやBigQueryでの集計結果をpandas DataFrameとして取得する方法について紹介しました。

明日は、@fiemonさんによる「Wikipediaデータ MySQLを経由してBigQueryにLoadする」です。
引き続き、GMOアドマーケティング Advent Calendar 2019 をお楽しみください！

■エンジニア採用ページ　～福利厚生や各種制度のご案内はこちら～
https://www.gmo-ap.jp/engineer/
■Wantedlyページ　～ブログや求人を公開中！～
https://www.wantedly.com/projects/199431
■エンジニア学生インターン募集中！　～就業型インターンでアドテクの先端技術を体験しよう～
https://hrmos.co/pages/gmo-ap/jobs/0000027

S.S

はじめに

外部データベースからのデータ取得(MySQL)

外部データベースからのデータ取得(BigQuery)

おすすめ

【Python】噂の”LightweightMMM”を使ってみた

GoogleのVertex AI PaLM Embedding APIを使って類似記事タイトルをレコメンドする

Slackアプリ経由でChatGPTと連想ゲームをしてみた。