ボットアクセス可視化のためのデータフロー構築

はじめに

本記事は昨今増加しているボットアクセスの可視化をするために必要な、
データ連携フロー構築に関するナレッジになります。

関連するクラウドサービスはAWSおよびGoogleになります。
担当しているサービスがAWS基盤であり、集計および可視化がGoogle基盤のため
データを加工して連携するロジックの共有となります。
※Google側に共有したあとの可視化までの流れは本記事のスコープ外とします。

同様の状況にあり、これから可視化するためのフローを構築する方に役立てば幸いです。
また、単にAWSとGCPのクロスクラウドデータ連携の必要がある方にも役立つかもしれません。

背景と状況

・Webサーバーのアクセスログを活用してアクセス状況を可視化したい
・サービス(元データ側)はS3にWebサーバー(Apache)のログを日次で退避している
※サーバーからS3へ退避されている前提として、その辺は割愛させていただきます
・Google側には既に可視化基盤が整っている(BigQuery+Looker Studio)

全体像

今回のデータフロー全体像を先に示します。

利用サービス

・AWS S3
・AWS Glue
・Google Storage Transfer Service
・Google Cloud Storage
・Google BigQuery
・Google Looker Studio

課題と解決策

データ加工

元データのApacheアクセスログと既存の可視化基盤の項目には差異があるため
AWS側で加工してすぐ可視化できる状態で引き渡す必要がありました。
→データ加工はGlueジョブのスクリプト内で実施しました。

ボット種類による分類

Google系ボットのアクセスか、それ以外のAIボットのアクセスか
によって分類して集計したい要望がありました。
とはいえApacheアクセスログには混在しているため、それらを切り分けて抽出する必要がありました。
→切り分けはGlueジョブ実行時のパラメータ(Glueトリガー側)で実施しました。

クロスクラウドデータ連携

BigQuery Data Transferが利用できず、Storage Transfer Serviceを初採用した点。
既存でナレッジのあった前者Data TransferではIAM ユーザーのアクセスキー、シークレットキーを利用するため
セキュリティ懸念があり、今回は採用しない方針としました。
→後者Storage Transfer ServiceではAssumeRoleを利用するためセキュリティリスクが下がります。
　既存ナレッジもフロー検討時に参考にさせていただきましたが、
　時代に合わせて実装も変えていく必要があるということですね。

実装

Glue ジョブスクリプト

既存のGlueジョブスクリプトでデータ連携系のものもありましたが、
今回S3格納済ログファイルのデータ加工が発生するため完全新規で作成しました。
スクリプトはほとんどAIで生成したため、参考までにプロンプトも添えておきます。

AIプロンプト

文章は重要ではないのでキャプチャで割愛します。
(BigQuery Data TransferはStorage Transfer Serviceに読み替えてください)

※実装当初は東京リージョン未対応だったようですが、2026年9月現在は対応されているようです。
https://docs.cloud.google.com/storage-transfer/docs/source-amazon-s3?hl=ja

...もろもろ壁打ちの結果、できた設計書が以下です。

簡易設計書

AIとのやり取り、最終結果どうなのかを簡単にまとめる習慣が個人的にあるのですが、
スクリプト生成の際などはアウトプットとして設計書を出しておくのが有意義だと思います。

# botアクセス可視化向け データ加工設計書（AWS Glue）

## 1. 概要
本設計書は、Apacheアクセスログを対象に、botアクセスのみを抽出・加工し、  
BIツール（BigQuery / Looker Studio）での可視化に適した形式で連携するための  
データ加工処理（AWS Glue）の設計方針を示すものである。

---

## 2. 全体構成

### 2.1 処理フロー
CloudFront / EC2 Apache Log
↓
S3（ap-northeast-1）
↓
AWS Glue（バッチ・日次）
↓
S3（us-east-1）
↓
Storage Transfer Service
↓
Cloud Storage
↓
BigQuery
↓
Looker Studio


### 2.2 担当範囲
- 本設計書の対象範囲  
  - S3（ログ）→ AWS Glue → S3（加工済データ）
- BIツール側（BigQuery / Looker Studio）はサイト運営チームの担当とする

---

## 3. インプット仕様

### 3.1 入力データ格納先
- リージョン：ap-northeast-1
- バケット：

s3://xxxxxxxxxxxxxxx/ec2-log/
└ prod-webap0X/apache.access/{yyyy}/{MM}/{dd}/{hh}/

### 3.2 ファイル形式
- テキスト（TSV）
- 列構成：
| 列番号 | 内容 |
|------|------|
| 1 | 収集時刻（ISO8601） |
| 2 | サーバ名 / ログ種別 |
| 3 | Apacheアクセスログ（JSON文字列） |

---

## 4. データ加工概要

### 4.1 処理単位
- 日次バッチ
- 前日分（yyyy/MM/dd）のみを対象に処理を行う

### 4.2 処理内容
1. Apacheログ(JSON)の構造化
2. user-agent による botアクセス判定
3. botアクセスのみ抽出
4. BI向けにフィールドを正規化
5. CSV形式で S3（us-east-1）へ出力

---

## 5. bot判定仕様

### 5.1 判定対象フィールド
- JSON内 `agent`（User-Agent）

### 5.2 判定方法
- user_agent を小文字化した上で、  
判定用トークンとの **部分一致または完全一致** により判定する
- いずれかのトークンに一致した場合、botアクセスとして採用する

### 5.3 判定用トークン一覧

OAI-SearchBot
ChatGPT-User
GPTBot
Claude-SearchBot
Claude-User
ClaudeBot
Claude-Web
anthropic-ai
Google-CloudVertexBot
Google-Extended
PerplexityBot
Perplexity-User
Applebot-Extended
meta-externalagent
Meta-WebIndexer
SBIntuitionsBot
MistralAI-User
cohere-ai
cohere-training-data-crawler
img2dataset
ChatGLM-Spider
AI2Bot
sider.ai

---

## 6. アウトプット仕様

### 6.1 出力先
- リージョン：us-east-1
- バケット：

s3://xxxxxxxxxxxxxxx/group-a/{yyyy}/{MM}/{dd}/

### 6.2 ファイル形式
- CSV（ヘッダあり）

---

## 7. 出力項目定義

出力する項目は以下に限定し、  
**項目名・順序ともに固定**とする。

| No | 項目名 | マッピング元 | 備考 |
|---|---|---|---|
| 1 | date | 収集時刻 | yyyy-MM-dd |
| 2 | time | 収集時刻 | HH:mm:ss |
| 3 | x-edge-location | - | NULL固定 |
| 4 | c-ip | host | 先頭IPのみ |
| 5 | cs-method | method | |
| 6 | cs(Host) | - | NULL固定 |
| 7 | cs-uri-stem | path | |
| 8 | sc-status | code | 数値 |
| 9 | cs(Referer) | referer | "-"はNULL |
|10 | cs(User-Agent) | agent | |
|11 | time-taken | - | NULL固定 |

---

## 8. 非機能要件

### 8.1 冪等性
- 同一日付の再実行時は出力データを上書きする

### 8.2 エラーハンドリング
- 一部サーバのログ欠損時でも処理を継続可能とする
- Glueジョブ失敗時は再実行可能な設計とする

---

## 9. 補足
- 本データは botアクセス可視化を目的とし、人間アクセスは出力対象外とする
- 後続処理（BigQuery / Looker Studio）での集計・分析を前提に、  
CloudFrontログ形式に近い項目構成としている

スクリプト: bot_access_to_s3

ほぼほぼAIの生成結果です。
要件伝えるための壁打ちはある程度必要でしたが、
コーディングに関してはほとんどお任せできました。

import sys
import re
import logging
from datetime import datetime, timedelta, timezone

import boto3
from pyspark.context import SparkContext
from pyspark.sql import functions as F
from pyspark.sql.types import (
    StructType, StructField,
    StringType
)

from awsglue.context import GlueContext
from awsglue.job import Job


# ----------------------------
# Optional args parser (GlueのgetResolvedOptionsは必須引数に弱いので自前)
# ----------------------------
def get_arg(name: str, default=None):
    """
    Glueの引数は --key value の形式で渡される前提
    未指定なら default を返す
    """
    key = f"--{name}"
    if key in sys.argv:
        idx = sys.argv.index(key)
        if idx + 1 < len(sys.argv):
            return sys.argv[idx + 1]
    return default


def parse_s3_uri(uri: str):
    """
    s3://bucket/prefix... を (bucket, prefix) に分解
    """
    if not uri.startswith("s3://"):
        raise ValueError(f"Invalid S3 URI: {uri}")
    no_scheme = uri[5:]
    parts = no_scheme.split("/", 1)
    bucket = parts[0]
    prefix = parts[1] if len(parts) > 1 else ""
    return bucket, prefix


def ensure_trailing_slash(s: str) -> str:
    return s if s.endswith("/") else s + "/"


# ----------------------------
# Normalize helpers
# ----------------------------
_HYPHENS = [
    "\u2010", "\u2011", "\u2012", "\u2013", "\u2014", "\u2212"
]
def normalize_hyphen(s: str) -> str:
    if s is None:
        return None
    for h in _HYPHENS:
        s = s.replace(h, "-")
    return s


# ----------------------------
# Logging
# ----------------------------
def setup_logger(level: str):
    logger = logging.getLogger("bot-access-glue")
    logger.setLevel(level.upper() if level else "INFO")
    handler = logging.StreamHandler(sys.stdout)
    fmt = logging.Formatter("%(asctime)s %(levelname)s %(message)s")
    handler.setFormatter(fmt)
    # Glueは二重にハンドラが付くことがあるのでガード
    if not logger.handlers:
        logger.addHandler(handler)
    return logger


# ----------------------------
# S3 prefix exists (boto3)
# ----------------------------
def s3_prefix_exists(s3_client, bucket: str, prefix: str) -> bool:
    """
    prefix配下に1件でもオブジェクトがあれば True
    """
    resp = s3_client.list_objects_v2(Bucket=bucket, Prefix=prefix, MaxKeys=1)
    return "Contents" in resp


# ----------------------------
# Main
# ----------------------------
def main():
    # ---- Parameters ----
    # 必須
    input_s3_base_uri = get_arg("input_s3_base_uri", None)   # 例: s3://xxxxxxxxxxxxxxx/ec2-log/
    output_s3_base_uri = get_arg("output_s3_base_uri", None) # 例: s3://xxxxxxxxxxxxxxx/group-a/
    target_token = get_arg("target_token", None)             # 例: "GPTBot,Claude-User,..."

    # 任意
    target_date = get_arg("target_date", None)               # "YYYY-MM-DD" / 未指定なら前日
    dry_run = (get_arg("dry_run", "false").lower() == "true")
    output_mode = get_arg("output_mode", "overwrite")        # overwrite | append
    log_level = get_arg("log_level", "INFO")                 # INFO | DEBUG
    max_server_count = int(get_arg("max_server_count", "8")) # デフォ8
    output_coalesce = int(get_arg("output_coalesce", "0"))   # 0:そのまま / 1:単一ファイル化（小規模向け）
    bot_type = get_arg("bot_type", "ai")                     # ai | google_mobile
    

    logger = setup_logger(log_level)

    # validate required
    if not input_s3_base_uri or not output_s3_base_uri or not target_token:
        raise ValueError(
            "Missing required args. "
            "Required: --input_s3_base_uri --output_s3_base_uri --target_token"
        )

    input_s3_base_uri = ensure_trailing_slash(input_s3_base_uri)
    output_s3_base_uri = ensure_trailing_slash(output_s3_base_uri)

    # target_date default: yesterday (JST)
    if not target_date:
        # Glue実行リージョンが東京想定なので JST(UTC+9)で「前日」を作る
        jst = timezone(timedelta(hours=9))
        target_date = (datetime.now(jst) - timedelta(days=1)).strftime("%Y-%m-%d")

    # parse date parts
    dt = datetime.strptime(target_date, "%Y-%m-%d")
    yyyy = dt.strftime("%Y")
    mm = dt.strftime("%m")
    dd = dt.strftime("%d")

    # parse tokens
    # カンマ区切り、前後空白除去、小文字化、ハイフン正規化
    raw_tokens = [t.strip() for t in target_token.split(",") if t.strip()]
    tokens = [normalize_hyphen(t).lower() for t in raw_tokens]

    logger.info(f"Job started. target_date={target_date} dry_run={dry_run} output_mode={output_mode}")
    logger.info(f"Bot token count={len(tokens)} max_server_count={max_server_count}")
    logger.info(f"bot_type={bot_type}")

    # ---- Discover existing input prefixes ----
    s3 = boto3.client("s3")
    in_bucket, in_base_prefix = parse_s3_uri(input_s3_base_uri)  # base_prefix例: "ec2-log/"
    in_base_prefix = ensure_trailing_slash(in_base_prefix)

    existing_paths = []
    existing_servers = []
    missing_servers = []

    # 想定命名: prod-webap01 ... prod-webap08
    for i in range(1, max_server_count + 1):
        server = f"prod-webap{i:02d}"
        # 入力の実際パス: s3://xxxxxxxxxxxxxxx/ec2-log/prod-webap01/apache.access/yyyy/MM/dd/hh/
        # Sparkで日単位取得したいので dd/ までを prefix にする（hh配下をまとめて拾う）
        prefix = f"{in_base_prefix}{server}/apache.access/{yyyy}/{mm}/{dd}/"

        if log_level.upper() == "DEBUG":
            logger.debug(f"Checking prefix s3://{in_bucket}/{prefix}")

        if s3_prefix_exists(s3, in_bucket, prefix):
            existing_paths.append(f"s3://{in_bucket}/{prefix}")
            existing_servers.append(server)
        else:
            missing_servers.append(server)

    if not existing_paths:
        logger.error("No input logs found for any server. job aborted.")
        raise Exception(f"No input logs found for target_date={target_date}")

    logger.info(f"Found input paths={len(existing_paths)} servers={existing_servers}")
    if missing_servers:
        logger.warning(f"No logs found for servers={missing_servers}")

    # ---- Glue / Spark init ----
    sc = SparkContext.getOrCreate()
    glueContext = GlueContext(sc)
    spark = glueContext.spark_session
    job = Job(glueContext)
    job.init("bot-access-glue-job", {})  # job nameはGlue側で上書きされてもOK

    # ---- Read TSV (3 columns) ----
    # 形式: collected_at \t source \t json_str
    df_raw = (
        spark.read
        .option("sep", "\t")
        .option("quote", "\u0000")   # quote無効化（JSON中にquoteがあっても壊れにくくする）
        .option("escape", "\u0000")  # escape無効化
        .option("recursiveFileLookup", "true")  # 対象パスの配下も再帰的にチェック
        .csv(existing_paths)
        .toDF("collected_at", "source", "json_str")
    )

    # ---- Parse JSON (from_json) ----
    json_schema = StructType([
        StructField("host", StringType(), True),
        StructField("user", StringType(), True),
        StructField("time", StringType(), True),
        StructField("method", StringType(), True),
        StructField("path", StringType(), True),
        StructField("code", StringType(), True),
        StructField("size", StringType(), True),
        StructField("referer", StringType(), True),
        StructField("agent", StringType(), True),
        StructField("proto", StringType(), True),
    ])

    df = df_raw.withColumn("log", F.from_json(F.col("json_str"), json_schema))

    # ---- Normalize UA + token matching ----
    # UA: 小文字化 + ハイフン正規化
    # Spark側で置換する（簡易に代表的なハイフン類を "-" に寄せる）
    ua = F.col("log.agent")
    ua_norm = F.lower(ua)
    for h in _HYPHENS:
        ua_norm = F.regexp_replace(ua_norm, re.escape(h), "-")

    df = df.withColumn("ua_norm", ua_norm)

    cond = F.lit(False)
    for t in tokens:
        cond = cond | F.col("ua_norm").contains(t)
    
    # Googleボットはtarget_tokenで指定した文字列"Googlebot"に加えて"Mobile"も部分一致するものとする。
    if bot_type == "google_mobile":
        cond = cond & F.col("ua_norm").contains("mobile")

    df_bot = df.filter(cond)

    # ---- Timestamp parse (collected_at優先) ----
    # 例: 2026-04-19T20:00:06+09:00
    ts = F.to_timestamp(F.col("collected_at"), "yyyy-MM-dd'T'HH:mm:ssXXX")
    df_bot = df_bot.withColumn("ts", ts)

    # ---- Map required output fields ----
    # c-ip: hostの先頭（"ip1, ip2" -> ip1）
    client_ip = F.when(
        F.col("log.host").isNull(),
        F.lit(None)
    ).otherwise(
        F.trim(F.split(F.col("log.host"), ",")[0])
    )

    # referer "-" は NULL
    referer = F.when((F.col("log.referer") == "-") | (F.col("log.referer") == ""), F.lit(None)) \
               .otherwise(F.col("log.referer"))

    df_out = df_bot.select(
        F.date_format(F.col("ts"), "yyyy-MM-dd").alias("date"),
        F.date_format(F.col("ts"), "HH:mm:ss").alias("time"),
        # F.lit(None).cast(StringType()).alias("x-edge-location"),
        client_ip.alias("c-ip"),
        F.col("log.method").alias("cs-method"),
        # F.lit(None).cast(StringType()).alias("cs(Host)"),
        F.col("log.path").alias("cs-uri-stem"),
        F.col("log.code").cast("int").alias("sc-status"),
        referer.alias("cs(Referer)"),
        F.col("log.agent").alias("cs(User-Agent)"),
        # F.lit(None).cast(StringType()).alias("time-taken"),
    )

    # ---- Metrics / counts (高コストなのでDEBUG or dry_run時だけ) ----
    if dry_run or log_level.upper() == "DEBUG":
        read_cnt = df_raw.count()
        bot_cnt = df_out.count()
        logger.info(f"Records read={read_cnt}")
        logger.info(f"Records matched(bot)={bot_cnt}")
        if bot_cnt == 0:
            logger.warning(f"No bot traffic found for target_date={target_date}")

    # ---- Output path ----
    out_bucket, out_base_prefix = parse_s3_uri(output_s3_base_uri)
    out_base_prefix = ensure_trailing_slash(out_base_prefix)
    # 要件: s3://.../group-a/{yyyy}/{MM}/{dd}/
    out_prefix = f"{out_base_prefix}{yyyy}/{mm}/{dd}/"
    output_path = f"s3://{out_bucket}/{out_prefix}"

    logger.info(f"Output path={output_path}")

    # ---- Write ----
    if dry_run:
        logger.info("dry_run=true -> skip writing output.")
    else:
        writer = df_out
        if output_coalesce == 1:
            writer = writer.coalesce(1)

        (
            writer.write
            .mode(output_mode)             # overwrite | append
            .option("header", True)        # ヘッダあり
            .csv(output_path)
        )

    logger.info("Job completed successfully")
    job.commit()


if __name__ == "__main__":
    main()

Glueトリガー設定(パラメータ定義)

トリガーで以下のパラメータを指定してジョブを実行します。
主にtarget_token、output_s3_base_uriを微修正して、
googleボット、aiボットそれぞれのパラメータをプリセットしておき実行しています。

target_date: 未指定だったらデフォルトで前日、指定あったら特定日付で実行もできるように
target_token: 対象botの追加修正にも対応できるように、カンマ区切りで複数入れた文字列
input_s3_base_uri: それ以降のサーバーや日付ごとのパスはスクリプト側で実装する想定
output_s3_base_uri: target_tokenと合わせて別のトークングループが追加されたときに使う想定
dry_run: false | true
output_mode: overwrite | append
log_level: DEBUG | INFO
bot_type: ai | google_mobile

googleボット用トリガー: daily_google_bot_access_to_s3

aiボット用トリガー: daily_ai_bot_access_to_s3

クロスクラウドデータ連携: AWS→Google

Google側(連携先)

Google側はStorage Transfer Serviceのマニュアル(参考記事①)や、
データ転送のデベロッパーブログ(参考記事②)を参考にしていただければと思います。

プロジェクト作成

こちらはスコープ外なので作成済を前提として割愛させていただきます。

Cloud Storage バケット作成

こちらはスコープ外なので作成済を前提として割愛させていただきます。

サブジェクトIDの取得

こちらが今回の連携において肝で、Google側で事前に確認しておく必要があります。
Subject ID取得は以下リンク(googleServiceAccounts.get - Try this method)から可能です。
https://cloud.google.com/storage-transfer/docs/reference/rest/v1/googleServiceAccounts/get
確認した結果をAWS側に連携します。

AWS側(連携元)

OIDC 設定(IDプロバイダ)

AWSと初めて連携する場合はこれも必要になります。
ここで対象者にサブジェクトIDを入れる必要があるので要注意です。
私はここで少しハマりました...

プロバイダ: accounts.google.com
対象者: Google側のSubject IDとsts.amazonaws.com

The fix is to add the audience you're using to the IdentityProvider.
引用: (参考記事③)

IAMポリシー: GoogleLinkPolicy

Roleで参照するポリシーです。
対象のS3に関する権限のみをつけています。

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AllowReadOnlyAccessForGoogleStorageTransfer",
      "Effect": "Allow",
      "Action": [
        "s3:GetObject",
        "s3:ListBucket"
      ],
      "Resource": [
        "arn:aws:s3:::xxxxxxxxxxxxxxx",
        "arn:aws:s3:::xxxxxxxxxxxxxxx/*"
      ]
    }
  ]
}

IAM Role 作成（信頼ポリシー）

連携・認証用のRoleです。
ここにももちろんサブジェクトIDを入れる必要があります。
上で作成したIDプロバイダのARNも紐づけます。

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {
                "Federated": "arn:aws:iam::xxxxxxxxxxxxxxx:oidc-provider/accounts.google.com"
            },
            "Action": "sts:AssumeRoleWithWebIdentity",
            "Condition": {
                "StringEquals": {
                    "accounts.google.com:sub": "xxxxxxxxxxxxxxx"
                }
            }
        }
    ]
}

Google側(連携先)

ここまでの設定がうまくいっていればあとは簡単、
Storage Transfer Service（STS）にてストレージ転送ジョブの構成を定義するだけです。
デベロッパーブログ(参考記事②)を参考に設定いただければと思います。

以上となります。
長くなりましたが最後まで読んでいただいた方、ありがとうございます。

---

別記事リンク

参考記事①: Googleドキュメント

Storage Transfer Service
https://docs.cloud.google.com/storage-transfer/docs/overview?hl=ja

参考記事②: Googleデベロッパーブログ

Storage Transfer Service を使用して AWS から GCP にデータを転送する
https://cloud.google.com/blog/ja/topics/developers-practitioners/transfer-data-aws-gcp-using-storage-transfer-service/

参考記事③: AWSナレッジセンター

How do I resolve the AWS STS AssumeRoleWithWebIdentity API call error "InvalidIdentityToken"?
https://repost.aws/knowledge-center/iam-sts-invalididentitytoken