スクレイピングをクラウド上で実行する

先日BeautifulSoupを使いWebスクレイピングに挑戦した。
今回は、それを定期的にスクレイプしそれをDBに保存することにした。また、クラウド上に全てを引っ越すことにした。GitHub

ローカルでPythonコードを走らせ、スクレイプしたデータをGoogleドライブに保存する

mount.py

# Mount Google Drive
from google.colab import drive
drive.mount('/content/drive')

スクレイプ結果のCSVをGoogleドライブに保存

定期的にスクレイプし、ファイルを保存するため、ファイル名に日時を入れる。

filename.py

import datetime
date = datetime.datetime.now().strftime("%Y%m%d_%H%M")
filename = f'/content/drive/My Drive/Nagoya_{date}.csv'

CSVをGCP SQLにアップロードする

GoogleドライブのCSVをGCP SQLにアップロードする方法を調べたところ、GoogleブログのGoogleColabをそのまま利用すると出来るとのことだったので、参考にした。

また、これでGCP SQLにデータを入れることが出来た。GCP SQLを最安値で利用する為にこちらの記事を参考にした。

authenticate.py

# 権限設定
def nagoya_authenticate(プロジェクトID): #GCPプロジェクトIDを入力
    # Authenticate user
    from google.colab import auth
    auth.authenticate_user()

    # gcloud設定
    !gcloud config set project <プロジェクトID>　#GCPプロジェクトID
    return

connect.py

def connect():
    # Cloud SQLクライアントに権限付与
    current_user = !gcloud auth list --filter=status:ACTIVE --format="value(account)"

    project_id = <プロジェクトID>: #GCPプロジェクトIDを入力
    !gcloud projects add-iam-policy-binding {project_id} \
        --member=user:{current_user[0]} \
        --role="roles/cloudsql.client"

    # Cloud SQL Admin APIを有効化
    !gcloud services enable sqladmin.googleapis.com  

    # Google Cloud リージョン及び Cloud SQL インスタンス名を設定
    region = <リージョン> #設定するGCPリージョンを入力
    instance_name = <インスタンス名> #設定するCloud SQLインスタンス名を入力

    # Cloud SQL インスタンスが既に存在するか確認
    database_version = !gcloud sql instances describe {instance_name} --format="value(databaseVersion)"
    if database_version[0].startswith("MYSQL"):
        print("既にMySQL Cloud SQL インスタンスが存在します")
    else:
        print("新しいCloud SQLインスタンスを作成します")
        password = input("'root' ユーザーで使用するパスワードを入力して下さい: ")
        !gcloud sql instances create {instance_name} --database-version=MYSQL_8_0 \
            --region={region} --cpu=1 --memory=4GB --root-password={password} \
            --database-flags=cloudsql_iam_authentication=On

    # データベースを作成
    !gcloud sql databases create <データベース名> --instance={instance_name}

    # データベース操作に使うユーザーを作成
    !gcloud sql users create <ユーザー名> \
        --instance={instance_name} \
        --password=<パスワード> #設定するパスワードを入力

    # ディペンデンシーをインストール
    import sys
    !{sys.executable} -m pip install cloud-sql-python-connector["pymysql"] SQLAlchemy==2.0.7

    # パラメータ初期設定
    INSTANCE_CONNECTION_NAME = f"{project_id}:{region}:{instance_name}" 
    print(f"インスタンス接続名: {INSTANCE_CONNECTION_NAME}")
    DB_USER = <ユーザー名> # 上記で入力したユーザー名
    DB_PASS = <パスワード> # 上記で入力したパスワード
    DB_NAME = <データベース名> # 上記で入力したデータベース名

    from google.cloud.sql.connector import Connector
    import sqlalchemy

    # 接続オブジェクト初期設定
    connector = Connector()


    # データベース接続オブジェクトを返すファンクション作成
    def getconn():
        conn = connector.connect(
            INSTANCE_CONNECTION_NAME,
            "pymysql",
            user=DB_USER,
            password=DB_PASS,
            db=DB_NAME
        )
        return conn

    # creator引数で接続オブジェクトファンクションへの接続プールを作る
    pool = sqlalchemy.create_engine(
        "mysql+pymysql://",
        creator=getconn,
    )

    # 接続プールに接続する
    with pool.connect() as db_conn:
        # データベースに<テーブル名>テーブルを作成する
        db_conn.execute(
            sqlalchemy.text(
                "CREATE TABLE IF NOT EXISTS <テーブル名> "
                "(<カラム名1>  VARCHAR(255) NOT NULL, "
                "<カラム名2>  VARCHAR(255) NOT NULL, <カラム名3> VARCHAR(255) NOT NULL, "
                "<カラム名4>  VARCHAR(255) NOT NULL, <カラム名5> VARCHAR(255) NOT NULL );"
            )
        )

        # コミット
        db_conn.commit()
        # <テーブル名>テーブルにデータを挿入
        insert_stmt = sqlalchemy.text(
                "INSERT INTO <テーブル名> (<カラム名1>, <カラム名2>, <カラム名3>, <カラム名4>, <カラム名5>) VALUES (:<カラム名1>, :<カラム名2>, :<カラム名3>, :<カラム名4>, :<カラム名5>)",
        )


        import csv
        with open(filename, 'r') as file:
            reader = csv.reader(file)
            for row in reader:
                db_conn.execute(insert_stmt, parameters={<カラム名1>:row[0], <カラム名2>:row[1], <カラム名3>:row[2], <カラム名4>:row[3], <カラム名5>:row[4]})

            # コミット
            db_conn.commit()

    # 接続オブジェクト削除
    connector.close()

    # SQL例
    code = "select count(*) from nagoya;"
    with pool.connect() as db_conn:
        print(db_conn.execute(sqlalchemy.text(code)).fetchall())
    
    return

GCPにてCronジョブを走らせる

こちらの記事を参考にGCP VertexIAにて、Jupyter notebookをスケジュールすることが出来そうだ。試しに簡単なスクリプトをスケジューリングし走らせると、Cloud Storage bucketに結果が出力されている。

GCPの課金もかかるので、今回はGCPでのデータ管理はこれ以上は進めず、ここまでのやり方の確認までとした。

最後に

小さな個人プロジェクトでは、GCPを使うと課金がされてしまい、あまり効率的なやり方ではないと感じた。大まかであるが、GCP上で定期的にコードを走らせ、データを保存、SQLに保存する方法が分かった点が良かったと思う。