🐕

Pythonから踏み台経由でDBにApacheログ登録した話

2024/07/27に公開

MySQL

まえがき

以前の記事のサーバーのアクセスログをPythonで抽出して見える化した話の続きで、DataframeでApacheログを見れるようにしたが、当時の現場にはPythonを使える方がそれほど多くなく、SQLを使える人は大勢いた。

今回はPythonのDataframeから、MySQLにテーブルを作ってSQLでApacheログを見れるようにした話。

環境

ローカルにPython環境があり、AWSのRDSにMySQLが構築してある。
RDSへのアクセスは踏み台サーバー経由でしかアクセスできないようになっている。

実装方針

前の段階で作った日付ごとのDataframeがあるため、それを活用して日付ごとにDataframeを読み取り、踏み台サーバー経由でRDSに接続して、Dataframeをテーブル化していく方針にした。

Dataframeじゃ1ヶ月分で10GB以上と結構巨大なファイルとなっており、SQLを叩くときにも巨大テーブルに対してSQLを叩くよりかは、月毎のテーブルにして月毎にテーブルを作ることにした。

実際の実装

Pythonにて踏み台経由でRDSに接続する
Dataframeを読み込んでRDSにデータとして書き込む
1ヶ月ごとにテーブルを分ける

テーブル名はログということがわかりやすく以下のようにした。
log_access_yyyymm

以下の状態でdataframeが保存してある状態で動くスクリプトとなっている。

スクリプトが置いてあるディレクトリにdataframeディレクトリが用意してある
dataframeディレクトリの中にdf_access_log_yyyymmdd.pickleのファイル名でdataframeが保存してある

from sshtunnel import SSHTunnelForwarder
from sqlalchemy import create_engine
import pandas as pd
from datetime import datetime
import glob

# アクセスログをdataframeにした後、dataframeをdatabaseに登録する

REMOTE_SERVER_IP = "踏み台サーバーのIPアドレス"
SSH_PKEY = "踏み台への鍵のパス"
SSH_USERNAME = "ユーザー名"

DB_HOST = "RDSのホスト名"
DB_USER = "RDSのユーザー名"
DB_PASSWORD = "RDSのパスワード"
DB_PORT = "ポート番号（通常は3306）"
DB_NAME = "DB名"

server = SSHTunnelForwarder(
    REMOTE_SERVER_IP,
    ssh_username=SSH_USERNAME,
    ssh_pkey=SSH_PKEY,
    remote_bind_address=(DB_HOST, DB_PORT)
)


def add_dataframe(file_path, datetime_file):
    # 踏み台接続開始
    server.start()

    url_sql = f'mysql+pymysql://{DB_USER}:{DB_PASSWORD}@127.0.0.1:{server.local_bind_port}/{DB_NAME}'
    engine = create_engine(url_sql)

    # データ読み込み
    df_log = pd.read_pickle(file_path)
    print("len(df_log):{}".format(len(df_log)))

    # 書き込み
    print("Write to database start")
    df_log.to_sql(
        name="log_access_{}".format(datetime.strftime(datetime_file, "%Y%m")),
        con=engine,
        if_exists="append",
    )
    print("Write to database end")

    # 踏み台接続終了
    server.close()


if __name__ == '__main__':

    datetime_targ_start = datetime(2024, 1, 1)
    datetime_targ_end = datetime(2024, 1, 31)
    print("datetime_targ_start:{}".format(datetime_targ_start))
    print("datetime_targ_end:{}".format(datetime_targ_end))

    str_path = "./dataframe/df_access_log_*.pickle"
    path_list = glob.glob(str_path)
    print("len(path_list):{}".format(len(path_list)))

    for file_path in path_list:
        name_file = file_path.split("/")[-1]
        name_file = name_file.replace(".pickle", "")
        datetime_file = datetime.strptime(name_file, "%Y%m%d")
        if datetime_file < datetime_targ_start:
            continue
        if datetime_file > datetime_targ_end:
            continue
        print(file_path)
        add_dataframe(file_path, datetime_file)

インストールしたパッケージは以下となった。

bcrypt==4.2.0
cffi==1.16.0
cryptography==43.0.0
numpy==2.0.1
pandas==2.2.2
paramiko==3.4.0
pycparser==2.22
PyNaCl==1.5.0
python-dateutil==2.9.0.post0
pytz==2024.1
six==1.16.0
SQLAlchemy==2.0.31
sshtunnel==0.4.0
typing_extensions==4.12.2
tzdata==2024.1

まえがき

環境

実装方針

実際の実装

Discussion