📣新登場！TiDB Cloud BYOC、TiDB Cloud Filesystem、PostgreSQL-compatible TiDB Cloud Starter がパブリックプレビュー中。このページは自動翻訳されたものです。原文はこちらからご覧ください。
AI
開発者
ベストプラクティス
API

Vector を使用した JSON ログの取り込み (Cloud)

このチュートリアルでは、ローカルでのログ生成をシミュレートし、Vector を使用して収集し、S3 に保存し、スケジュールされたタスクを使用して TiDB Cloud Lake への取り込みを自動化します。

Vector による JSON ログのロード自動化

始める前に

開始する前に、次の前提条件を満たしていることを確認してください。

  • Amazon S3 Bucket: Vector が収集したログを保存する S3 バケット。S3 バケットの作成方法を参照してください。
  • AWS Credentials: S3 バケットにアクセスするための十分な権限を持つ AWS Access Key ID と Secret Access Key。AWS 認証情報の管理を参照してください。
  • AWS CLI: AWS CLI がインストールされており、S3 バケットにアクセスするために必要な権限で設定されていることを確認してください。
  • Docker: Vector のセットアップに使用するため、ローカルマシンに Docker がインストールされていることを確認してください。

ステップ 1: S3 バケットに対象フォルダを作成する

Vector が収集したログを保存するために、S3 バケットに logs という名前のフォルダを作成します。このチュートリアルでは、対象の保存先として s3://lake-doc/logs/ を使用します。

このコマンドは、lake-doc バケットに logs という名前の空のフォルダを作成します。

aws s3api put-object --bucket lake-doc --key logs/

ステップ 2: ローカルログファイルを作成する

ローカルログファイルを作成して、ログ生成をシミュレートします。このチュートリアルでは、ファイルパスとして /Users/eric/Documents/logs/app.log を使用します。

サンプルのログイベントを表すために、次の JSON Lines をファイルに追加します。

{"user_id": 1, "event": "login", "timestamp": "2024-12-08T10:00:00Z"}
{"user_id": 2, "event": "purchase", "timestamp": "2024-12-08T10:05:00Z"}

ステップ 3: Vector を設定して実行する

  1. ローカルマシンに vector.yaml という名前の Vector 設定ファイルを作成します。このチュートリアルでは、/Users/eric/Documents/vector.yaml に次の内容で作成します。

    sources:
  logs:
    type: file
    include:
      - "/logs/app.log"
    read_from: beginning

transforms:
  extract_message:
    type: remap
    inputs:
      - "logs"
    source: |
      . = parse_json(.message) ?? {}

sinks:
  s3:
    type: aws_s3
    inputs:
      - "extract_message"
    bucket: lake-doc
    region: us-east-2
    key_prefix: "logs/"
    content_type: "text/plain"
    encoding:
      codec: "native_json"
    auth:
      access_key_id: "<your-access-key-id>"
      secret_access_key: "<your-secret-access-key>"

  2. 設定ファイルとローカルのログディレクトリをマウントして、Docker を使用して Vector を起動します。

    docker run \
  -d \
  -v /Users/eric/Documents/vector.yaml:/etc/vector/vector.yaml:ro \
  -v /Users/eric/Documents/logs:/logs \
  -p 8686:8686 \
  --name vector \
  timberio/vector:nightly-alpine

  3. 少し待ってから、ログが S3 上の logs フォルダに同期されているか確認します。

aws s3 ls s3://lake-doc/logs/

ログファイルが S3 に正常に同期されていれば、次のような出力が表示されます。

2024-12-10 15:22:13          0
2024-12-10 17:52:42        112 1733871161-7b89e50a-6eb4-4531-8479-dd46981e4674.log.gz

これで、同期されたログファイルをバケットからダウンロードできます。

aws s3 cp s3://lake-doc/logs/1733871161-7b89e50a-6eb4-4531-8479-dd46981e4674.log.gz ~/Documents/

元のログと比較すると、同期されたログは NDJSON 形式になっており、各レコードは外側の log フィールドでラップされています。

{"log":{"event":"login","timestamp":"2024-12-08T10:00:00Z","user_id":1}}
{"log":{"event":"purchase","timestamp":"2024-12-08T10:05:00Z","user_id":2}}

ステップ 4: TiDB Cloud Lake でタスクを作成する

  1. Worksheet を開き、バケット内の logs フォルダにリンクする外部 stage を作成します。

    CREATE STAGE mylog 's3://lake-doc/logs/' CONNECTION=(
    ACCESS_KEY_ID = '<your-access-key-id>',
    SECRET_ACCESS_KEY = '<your-secret-access-key>'
);

    stage の作成に成功すると、その中のファイルを一覧表示できます。

    LIST @mylog;

┌─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┐
│                          name                          │  size  │                 md5                │         last_modified         │      creator     │
├────────────────────────────────────────────────────────┼────────┼────────────────────────────────────┼───────────────────────────────┼──────────────────┤
│ 1733871161-7b89e50a-6eb4-4531-8479-dd46981e4674.log.gz │    112 │ "231ddcc590222bfaabd296b151154844" │ 2024-12-10 22:52:42.000 +0000 │ NULL             │
└─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┘

  2. ログ内のフィールドに対応するカラムを持つテーブルを作成します。

    CREATE TABLE logs (
    event String,
    timestamp Timestamp,
    user_id Int32
);

  3. 外部 stage から logs テーブルにログをロードするスケジュールタスクを作成します。

    CREATE TASK IF NOT EXISTS myvectortask
    WAREHOUSE = 'eric'
    SCHEDULE = 1 MINUTE
    SUSPEND_TASK_AFTER_NUM_FAILURES = 3
AS
COPY INTO logs
FROM (
    SELECT $1:log:event, $1:log:timestamp, $1:log:user_id
    FROM @mylog/
)
FILE_FORMAT = (TYPE = NDJSON, COMPRESSION = AUTO)
MAX_FILES = 10000
PURGE = TRUE;

  4. タスクを開始します。

ALTER TASK myvectortask RESUME;

少し待ってから、ログがテーブルにロードされているか確認します。

SELECT * FROM logs;

┌──────────────────────────────────────────────────────────┐
│       event      │      timestamp      │     user_id     │
├──────────────────┼─────────────────────┼─────────────────┤
│ login            │ 2024-12-08 10:00:00 │               1 │
│ purchase         │ 2024-12-08 10:05:00 │               2 │
└──────────────────────────────────────────────────────────┘

この時点で LIST @mylog; を実行すると、ファイルは何も表示されません。これは、タスクが PURGE = TRUE で設定されており、ログのロード後に同期済みファイルが S3 から削除されるためです。

次に、ローカルのログファイル app.log にさらに 2 件のログを生成することをシミュレートします。

echo '{"user_id": 3, "event": "logout", "timestamp": "2024-12-08T10:10:00Z"}' >> /Users/eric/Documents/logs/app.log
echo '{"user_id": 4, "event": "login", "timestamp": "2024-12-08T10:15:00Z"}' >> /Users/eric/Documents/logs/app.log

ログが S3 に同期されるまで少し待ちます（logs フォルダに新しいファイルが表示されるはずです）。その後、スケジュールタスクによって新しいログがテーブルにロードされます。再度テーブルをクエリすると、次のログを確認できます。

SELECT * FROM logs;

┌──────────────────────────────────────────────────────────┐
│       event      │      timestamp      │     user_id     │
├──────────────────┼─────────────────────┼─────────────────┤
│ logout           │ 2024-12-08 10:10:00 │               3 │
│ login            │ 2024-12-08 10:15:00 │               4 │
│ login            │ 2024-12-08 10:00:00 │               1 │
│ purchase         │ 2024-12-08 10:05:00 │               2 │
└──────────────────────────────────────────────────────────┘

このページは役に立ちましたか？