FileLog テーブルエンジン

• ログファイルをサブスクライブする。
• サブスクライブしたログファイルに追記された新しいレコードを処理する。

​

テーブルの作成

CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster]
(
    name1 [type1] [DEFAULT|MATERIALIZED|ALIAS expr1],
    name2 [type2] [DEFAULT|MATERIALIZED|ALIAS expr2],
    ...
) ENGINE = FileLog('path_to_logs', 'format_name') SETTINGS
    [poll_timeout_ms = 0,]
    [poll_max_batch_size = 0,]
    [max_block_size = 0,]
    [max_threads = 0,]
    [poll_directory_watch_events_backoff_init = 500,]
    [poll_directory_watch_events_backoff_max = 32000,]
    [poll_directory_watch_events_backoff_factor = 2,]
    [handle_error_mode = 'default']

• path_to_logs – サブスクライブするログファイルのパス。ログファイルを含むディレクトリへのパス、または単一のログファイルへのパスを指定できます。相対パスは、file テーブル関数と同様に、user_files_path ディレクトリを基準に解決されます。ただし、サーバーの作業ディレクトリを基準にすでに user_files_path 内を指している相対パス (たとえば、公式 Docker イメージのようにサーバーがデータディレクトリ内で実行されている場合の user_files/my_app/app.log) は、作業ディレクトリを基準とするパスとして扱われます。パス内のファイル名には glob (*、?、{abc,def}、{N..M}) を使用でき、ディレクトリ内で一致するファイルのみを読み込めます。詳しくは glob によるファイルの選択 を参照してください。なお、ClickHouse で指定できるのは user_files ディレクトリ内のパスのみです。
• format_name - レコードのフォーマット。FileLog はファイル内の各行を個別のレコードとして処理するため、すべてのデータフォーマットがこれに適しているわけではありません。

• poll_timeout_ms - ログファイルから 1 回 poll する際のタイムアウト。デフォルト: stream_poll_timeout_ms。
• poll_max_batch_size — 1 回の poll で取得するレコードの最大数。デフォルト: max_block_size。
• max_block_size — poll の最大バッチサイズ (レコード数) 。デフォルト: max_insert_block_size。
• max_threads - ファイルをパースする最大スレッド数。デフォルトは 0 で、この場合は max(1, physical_cpu_cores / 4) が使用されます。
• poll_directory_watch_events_backoff_init - ディレクトリ監視スレッドの初期スリープ値。デフォルト: 500。
• poll_directory_watch_events_backoff_max - ディレクトリ監視スレッドの最大スリープ値。デフォルト: 32000。
• poll_directory_watch_events_backoff_factor - バックオフの進み方。デフォルトは指数関数的です。デフォルト: 2。
• handle_error_mode — FileLog エンジン のエラー処理方法。設定可能な値: default (レコードのパースに失敗した場合、直接の SELECT では例外がスローされます。テーブルが materialized view にストリーミングしている間は、そのレコードはスキップされ、エラーはサーバーログに書き込まれます) 、stream (例外メッセージと生のレコードが仮想カラム _error および _raw_record に保存されます) 。

​

説明

1. エンジンを使用して FileLog テーブルを作成し、それをデータストリームとして扱います。
2. 必要な structure を持つテーブルを作成します。
3. エンジンからのデータを変換し、あらかじめ作成したテーブルに格納する materialized view を作成します。

CREATE TABLE logs (
    timestamp UInt64,
    level String,
    message String
  ) ENGINE = FileLog('my_app/app.log', 'JSONEachRow');

CREATE TABLE daily (
    day Date,
    level String,
    total UInt64
  ) ENGINE = SummingMergeTree
  PARTITION BY toYYYYMM(day)
  ORDER BY (day, level);

CREATE MATERIALIZED VIEW consumer TO daily
    AS SELECT toDate(toDateTime(timestamp)) AS day, level, count() AS total
    FROM logs GROUP BY day, level;

SELECT level, sum(total) FROM daily GROUP BY level;

DETACH TABLE consumer;
ATTACH TABLE consumer;

​

glob によるファイルの選択

app.log         the file the application writes
app.log.1       the previous file, renamed by logrotate, compressed on the next rotation
app.log.2.gz    older files, compressed

​

仮想カラム

• _filename - ログファイル名。データ型: LowCardinality(String).
• _offset - ログファイル内のオフセット。データ型: UInt64.

• _raw_record - 正常にパースできなかった生のレコード。データ型: Nullable(String).
• _error - パース失敗時に発生した例外メッセージ。データ型: Nullable(String).

​

データの永続性