ログの保存と分析

ログは、イベントの主体、時刻、場所、内容などの重要な情報を含むシステム動作の詳細な記録です。運用の可観測性、ネットワークセキュリティ監視、ビジネス分析のニーズに駆動され、企業は一般的に分散したログを一元的に収集、保存、分析し、膨大なログ量から価値を抽出する必要があります。

このシナリオに対して、Apache Dorisは一般的なOLAP機能に加えて、逆転インデックスと高速全文検索を追加し、書き込みパフォーマンスとストレージフットプリントを限界まで押し上げます。Apache Dorisを使用することで、オープンで高性能、低コスト、統合されたログストレージおよび分析プラットフォームを構築できます。

本記事では、このソリューションに関する以下のトピックを扱います：

全体アーキテクチャ: Apache Doris上に構築されたログプラットフォームのコアコンポーネントと基盤アーキテクチャ。
機能と利点: Elasticsearchと比較した差別化された機能。
運用ガイド: リソース見積もりからログクエリまでのエンドツーエンドの手順。

1. 全体アーキテクチャ

Apache Doris上に構築されたログストレージおよび分析プラットフォームのアーキテクチャを以下に示します：

Overall architecture

全体アーキテクチャは3つの部分で構成されています：

レイヤー	コンポーネント	説明
ログ収集と前処理	Logstash, Filebeat, Fluentbit, Kafka, など	HTTP APIを通じてApache Dorisにログデータを書き込み
ログストレージと分析	Apache Doris	高性能、低コストの統合ストレージと、SQLインターフェースを通じた豊富な検索・分析機能を提供
ログ分析とアラート	Grafana, Superset, Doris WebUI, など	標準MySQLプロトコルを通じてDorisにクエリを実行し、使いやすいビジュアルインターフェースを提供

2. 機能と利点

Apache Doris上に構築されたログプラットフォームは、以下のコア機能を提供します：

機能	説明
高スループット、低レイテンシインジェスト	1日あたり数百TBの安定した継続的なインジェストをGB/sでサポート、1秒未満のレイテンシ
大容量データの低コストストレージ	PBスケールのストレージをサポート、Elasticsearchと比較して60%–80%のストレージコストを削減、S3/HDFSへのコールドデータ階層化でさらに50%削減
高性能全文検索と分析	逆転インデックスと全文検索をサポート；キーワード詳細検索やトレンド分析などのクエリが秒単位で応答
オープンで使いやすいエコシステム	上流でLogstash, Filebeat, Fluentbit, Kafkaなどに接続；下流で標準MySQLプロトコルを通じてGrafana, Superset, Doris WebUIに接続

2.1 高性能と低コスト

ベンチマークと本番検証を通じて、Apache Doris上に構築されたログプラットフォームのコストパフォーマンス比は、Elasticsearchの5〜10倍優れています。この利点は主に高性能ストレージおよびクエリエンジンと、ログシナリオに特化した最適化から生まれています：

より高い書き込みスループット: Elasticsearchの書き込みボトルネックは、データ解析と逆転インデックス構築で消費されるCPUです。Apache DorisはSIMDなどのCPUベクトル化命令を使用してJSON解析とインデックス構築を高速化し、ログシナリオで不要な前方インデックスやその他のデータ構造を削除することで逆転インデックス構造を簡素化します。同じリソースで、Dorisの書き込みパフォーマンスはElasticsearchの3〜5倍です。
より低いストレージコスト: Elasticsearchは複数のデータコピー（前方インデックス、逆転インデックス、doc valuesカラムストア）と比較的低い一般的な圧縮率に悩まされます。Dorisは前方インデックスを削除してインデックスデータを30%削減；Zstandard圧縮アルゴリズムを使用したカラムナーストレージを採用し、Elasticsearchの1.5を大幅に上回る5〜10の圧縮率を実現します。ホット・コールド階層化機能により、履歴ログを自動的にオブジェクトストレージに移動し、コールドデータのストレージコストを70%以上削減できます。全体的なストレージコストはElasticsearchの約20%のみです。
より高いクエリパフォーマンス: Dorisは全文検索パイプラインを簡素化し、ログシナリオで不要な関連性スコアリングなどのアルゴリズムをスキップします。「指定キーワードを含む最新100件のログ」などの典型的なクエリに対して、DorisはプランニングおよびI実行レイヤーでTopN動的プルーニングなどの専用最適化を適用します。

2.2 強力な分析機能

Apache Dorisは標準SQLをサポートし、MySQLプロトコルと構文に互換性があるため、Doris上に構築されたログシステムには以下の利点があります：

使いやすさ: エンジニアとデータアナリストはSQLに高度に慣れ親しんでおり、新しい技術スタックを学習することなく迅速に開始できます。
豊富なエコシステム: MySQLコマンドライン、各種GUI / BIツール、ビッグデータエコシステムとシームレスに統合し、複雑で多様なデータ処理および分析ニーズを満たします。
強力な分析力: SQLはデータ分析のデファクトスタンダードであり、検索、集約、複数テーブルJOIN、サブクエリ、UDF、論理ビュー、マテリアライズドビューなどをサポートします。

2.3 柔軟なスキーマ

以下は、JSON形式の半構造化ログの典型的なサンプルです。トップレベルフィールドは比較的固定されており（timestamp, source, node, component, level, clientRequestId, message, properties）、拡張属性properties内のネストしたフィールド（properties.sizeやproperties.formatなど）はより動的で、フィールドはログごとに異なる場合があります。

{
  "timestamp": "2014-03-08T00:50:03.8432810Z",
  "source": "ADOPTIONCUSTOMERS81",
  "node": "Engine000000000405",
  "level": "Information",
  "component": "DOWNLOADER",
  "clientRequestId": "671db15d-abad-94f6-dd93-b3a2e6000672",
  "message": "Downloading file path: benchmark/2014/ADOPTIONCUSTOMERS81_94_0.parquet.gz",
  "properties": {
    "size": 1495636750,
    "format": "parquet",
    "rowCount": 855138,
    "downloadDuration": "00:01:58.3520561"
  }
}

Apache Dorisは以下の2つのメカニズムにより柔軟なスキーマをサポートしています：

Light Schema Change: トップレベルフィールドが時々変更される場合、ADD/DROP COLUMNやADD/DROP INDEXなどのスキーマ変更を数秒で完了できます。計画段階では、現在どのフィールドにインデックスが必要かを考慮するだけで済みます。
VARIANT半構造化タイプ: propertiesなどの拡張フィールドについて、任意のJSONデータを書き込むことができ、フィールド名と型は自動的に検出され、頻繁に出現するフィールドは列指向ストレージに分割されます。VARIANTに転置インデックスを作成して、内部フィールドのクエリと検索を高速化することも可能です。

Elasticsearchの動的マッピングと比較して、Apache Dorisの柔軟なスキーマには以下の利点があります：

同一フィールドに対する複数の型を許可します。VARIANTは自動的に競合を処理し型を昇格させ、ログデータの反復的変更により適応します。
VARIANTは使用頻度の低いフィールドを自動的に単一の列にマージし、フィールド、メタデータ、または列が多すぎることによるパフォーマンス問題を回避します。
動的な列の追加と削除、動的なインデックスの追加と削除をサポートするため、すべてのフィールドに対して事前にインデックスを作成する必要がなく、不要なコストを削減します。

3. 運用ガイド

以下の表は、Apache Doris上にログプラットフォームを構築するためのエンドツーエンドの6ステッププロセスを示しています：

ステップ	目的
ステップ1: リソースの見積もり	FE/BEノード数、ディスク容量、オブジェクトストレージサイズの見積もり
ステップ2: クラスタのデプロイ	物理マシンまたは仮想マシンへのApache Dorisのデプロイ
ステップ3: FEとBE設定の調整	ログシナリオ向けの主要パラメータの調整
ステップ4: テーブルの作成	パーティショニングとバケット化、圧縮、コンパクション、インデックス、階層化ポリシーの設計
ステップ5: ログの収集	Logstash、Filebeat、Kafka、またはカスタムプログラムとの統合
ステップ6: ログのクエリと分析	SQLと可視化ツールを使用したログの検索と分析

3.1 リソースの見積もり

クラスタをデプロイする前に、サーバハードウェアリソースを見積もる必要があります。主要なステップは以下の通りです：

書き込みリソースの見積もりを以下の公式で行います：
- 平均書き込みスループット = 日次増分データ / 86400秒
- ピーク書き込みスループット = 平均書き込みスループット * ピーク・平均書き込み比
- ピーク書き込みに必要なCPUコア数 = ピーク書き込みスループット / コア当たり書き込みスループット
ストレージリソースの見積もりを以下の公式で行います：
- 必要ストレージ = 日次増分データ / 圧縮率 * レプリカ数 * データ保持期間
クエリリソースの見積もり: クエリリソース消費量はクエリ量と複雑さによって異なります。初期予算として、CPUリソースの50%をクエリ用に確保し、実際のテストに基づいて調整します。
リソースの集約: ステップ1と3から必要なCPUコア数を見積もり、マシン当たりCPUコア数で除算してBEサーバ数を求めます。これをステップ2と組み合わせてBEサーバ当たりの必要ストレージを見積もり、4～12個のデータディスクに分散してディスク当たり容量を計算します。

例: 1日100TBの新規ログのリソース見積もり

以下の条件を例に取ります：1日100TBの新規データ（圧縮前）、圧縮率5、1レプリカ、ホットデータ3日間保持、コールドデータ30日間保持、ピーク・平均書き込み比200%、コア当たり書き込みスループット10MB/s、CPUの50%をクエリ用に確保。見積もりは以下の通りです：

FE: 3台のサーバ、各々16CPUコア、64GBメモリ、100GB SSD 1台
BE: 15台のサーバ、各々32CPUコア、256GBメモリ、600GB SSD 10台
S3オブジェクトストレージ: コールドデータストレージ容量、600TB

主要メトリクスの値と計算は以下の通りです：

主要メトリクス（単位）	値	説明
日次増分データ（TB）	100	実際のニーズに基づいて入力
圧縮率	5	通常3～10（インデックス含む）；実際のニーズに基づいて入力
レプリカ数	1	実際のニーズに基づいて入力；デフォルト1；可能な値：1、2、3
ホットデータ保持期間（日）	3	実際のニーズに基づいて入力
コールドデータ保持期間（日）	30	実際のニーズに基づいて入力
総保持期間（日）	33	公式：`ホットデータ保持期間 + コールドデータ保持期間`
ホットデータストレージ見積もり（TB）	60	公式：`日次増分データ / 圧縮率 * レプリカ数 * ホットデータ保持期間`
コールドデータストレージ見積もり（TB）	600	公式：`日次増分データ / 圧縮率 * レプリカ数 * コールドデータ保持期間`
ピーク・平均書き込み比	200%	実際のニーズに基づいて入力；デフォルト200%
マシン当たりCPUコア数	32	実際のニーズに基づいて入力；デフォルト32コア
平均書き込みスループット（MB/s）	1214	公式：`日次増分データ / 86400秒`
ピーク書き込みスループット（MB/s）	2427	公式：`平均書き込みスループット * ピーク・平均書き込み比`
ピーク書き込みに必要なCPUコア数	242.7	公式：`ピーク書き込みスループット / コア当たり書き込みスループット`
クエリ用に確保するCPU	50%	実際のニーズに基づいて入力；デフォルト50%
BEサーバ数見積もり	15.2	公式：`ピーク書き込みに必要なCPUコア数 / マシン当たりCPUコア数 / (1 - クエリ用に確保するCPU)`
BEサーバ数見積もり（四捨五入）	15	公式：`MAX(レプリカ数, 四捨五入したBEサーバ数見積もり)`
BEサーバ当たりストレージ見積もり（TB）	5.7	公式：`ホットデータストレージ見積もり / BEサーバ数見積もり / (1 - 30%)`、ここで30%はストレージ予備容量。I/O容量を向上させるため、BE当たり4～12個のデータディスクをマウントすることを推奨

3.2 クラスタのデプロイ

リソース見積もり後、Apache Dorisクラスタをデプロイできます。物理マシンまたは仮想マシンへのデプロイを推奨します。手動デプロイ手順については、手動デプロイを参照してください。

3.3 FEとBE設定の調整

クラスタデプロイ後、FEとBEパラメータを個別に調整して、ログストレージと分析シナリオにより適合させます。

3.3.1 FE設定の調整

fe/conf/fe.confで、以下の表に従ってFE設定を調整します：

調整するパラメータ	説明
`max_running_txn_num_per_db = 10000`	高並行取り込みは実行中トランザクションが多いため、このパラメータを引き上げ
`streaming_label_keep_max_second = 3600` `label_keep_max_second = 7200`	頻繁な取り込みトランザクションラベルはより多くのメモリを消費するため、保持時間を短縮
`enable_round_robin_create_tablet = true`	タブレット作成時にラウンドロビンを使用して、分散を可能な限り均等に保つ
`tablet_rebalancer_type = partition`	バランシング時に各パーティション内でタブレットを可能な限り均等に保つ戦略を使用
`autobucket_min_buckets = 10`	ログ量が増加した際にバケット不足を避けるため、自動バケット化の最小カウントを1から10に引き上げ
`max_backend_heartbeat_failure_tolerance_count = 10`	ログシナリオではBEサーバが高負荷でハートビートを一時的に逃すことがあるため、許容回数を1から10に引き上げ

より多くのパラメータ情報については、FE設定を参照してください。

3.3.2 BE設定の調整

be/conf/be.confで、以下の表に従ってBE設定を調整します：

モジュール	調整するパラメータ	説明
Storage	`storage_root_path = /path/to/dir1;/path/to/dir2;...;/path/to/dir12`	ディスクディレクトリ上のホットデータ用ストレージパスを設定
Storage	`enable_file_cache = true`	ファイルキャッシュを有効化
Storage	`file_cache_path = [{"path": "/mnt/datadisk0/file_cache", "total_size":53687091200, "query_limit": "10737418240"},{"path": "/mnt/datadisk1/file_cache", "total_size":53687091200,"query_limit": "10737418240"}]`	コールドデータ用のキャッシュパスと設定を構成： `path`: キャッシュパス `total_size`: キャッシュパスの総サイズ（バイト単位）；53687091200バイト = 50GB `query_limit`: 単一クエリがキャッシュパスから読み込める最大データ（バイト単位）；10737418240バイト = 10GB
Write	`write_buffer_size = 1073741824`	書き込みバッファファイルサイズを増加して小ファイルとランダムI/Oを削減し、パフォーマンスを向上
Compaction	`max_cumu_compaction_threads = 8`	CPUコア数 / 4に設定、つまりCPUの1/4を書き込み用、1/4をバックグラウンドコンパクション用、1/2をクエリと他の操作用に確保
Compaction	`inverted_index_compaction_enable = true`	インデックスコンパクションを有効化して、コンパクション時のCPU消費を削減
Compaction	`enable_segcompaction = false` `enable_ordered_data_compaction = false`	ログシナリオで不要な2つのコンパクション機能を無効化
Compaction	`enable_compaction_priority_scheduling = false`	低優先度コンパクションはディスク当たり2タスクに制限され、コンパクション速度を低下させる
Compaction	`total_permits_for_compaction_score = 200000`	メモリ制御に使用；時系列ポリシー自体でメモリを制御可能
Cache	`disable_storage_page_cache = true` `inverted_index_searcher_cache_limit = 30%`	ログデータ量が大きいため、データキャッシュの恩恵は限定的；データキャッシュを無効化してインデックスキャッシュを使用
Cache	`inverted_index_cache_stale_sweep_time_sec = 3600` `index_cache_entry_stay_time_after_lookup_s = 3600`	インデックスキャッシュを最長1時間メモリに保持
Cache	`enable_inverted_index_cache_on_cooldown = true` `enable_write_index_searcher_cache = false`	コールドストレージへのアップロード時にインデックスの自動キャッシュを有効化
Cache	`tablet_schema_cache_recycle_interval = 3600` `segment_cache_capacity = 20000`	他のキャッシュが使用するメモリを削減
Cache	`inverted_index_ram_dir_enable = true`	書き込み時にインデックス一時ファイルによるI/Oオーバーヘッドを削減
Threads	`pipeline_executor_size = 24` `doris_scanner_thread_pool_thread_num = 48`	32コアCPU用の計算とI/Oスレッド設定；コア数に基づいて比例的にスケールアップまたはダウン
Threads	`scan_thread_nice_value = 5`	クエリI/Oスレッドの優先度を下げ、書き込みパフォーマンスと適時性を確保
Other	`string_type_length_soft_limit_bytes = 10485760`	`String`データの長さ制限を10MBに引き上げ
Other	`trash_file_expire_time_sec = 300` `path_gc_check_interval_second = 900` `path_scan_interval_second = 900`	トラッシュファイルのリサイクルを高速化

より多くのパラメータ情報については、BE設定を参照してください。

3.4 テーブルの作成

ログデータの書き込みとクエリには両方とも独特の特性があるため、パフォーマンスを向上させるために、テーブル作成時には以下のセクションの対象設定に従ってください。

3.4.1 パーティショニングとバケット化パラメータの設定

パーティショニング:

時間フィールドでのRange partitioningを使用し（PARTITION BY RANGE(ts)）、Dynamic partitioningを有効化して（"dynamic_partition.enable" = "true"）、日次パーティションを自動管理します。
Datetime時間フィールドをキーとして使用します（DUPLICATE KEY(ts)）；これにより最新のNログをクエリする際に複数倍の高速化を提供します。

バケット化:

バケット数をクラスタ内の総ディスク数の約3倍に設定し、バケット当たり約5GBの圧縮データとします。
ランダム戦略を使用し（DISTRIBUTED BY RANDOM BUCKETS 60）、書き込み時に単一タブレット取り込みと組み合わせて、バッチ書き込み効率を向上させます。

パーティショニングとバケット化の詳細情報については、Data partitioningを参照してください。

3.4.2 圧縮パラメータの設定

Zstd圧縮アルゴリズムを使用して（"compression" = "zstd"）、データ圧縮率を向上させます。

3.4.3 コンパクションパラメータの設定

時系列ポリシーを使用して（"compaction_policy" = "time_series"）、書き込み増幅を軽減します。これは高スループットログ書き込みのリソース消費にとって重要です。

3.4.4 インデックスの構築と設定

頻繁にクエリされるフィールドにインデックスを構築します（USING INVERTED）。
全文検索が必要なフィールドには、トークナイザー（parser）パラメータをunicodeに設定します。これはほとんどのニーズに適合します。フレーズクエリをサポートするにはsupport_phraseをtrueに設定し、不要な場合はfalseに設定してストレージフットプリントを削減します。

3.4.5 ストレージポリシーの設定

ホットデータストレージ: クラウドディスク使用時は1レプリカを設定可能；物理ディスク使用時は最低2レプリカを設定します（"replication_num" = "2"）。
ホット・コールド階層化: log_s3ストレージ場所を設定し（CREATE RESOURCE "log_s3"）、log_policy_3day階層化ポリシーを設定して（CREATE STORAGE POLICY log_policy_3day）、3日より古いデータがlog_s3で指定されたストレージ場所に自動的にコールド化されるようにします。

3.4.6 完全なテーブル作成例

CREATE DATABASE log_db;
USE log_db;

CREATE RESOURCE "log_s3"
PROPERTIES
(
    "type" = "s3",
    "s3.endpoint" = "your_endpoint_url",
    "s3.region" = "your_region",
    "s3.bucket" = "your_bucket",
    "s3.root.path" = "your_path",
    "s3.access_key" = "your_ak",
    "s3.secret_key" = "your_sk"
);

CREATE STORAGE POLICY log_policy_3day
PROPERTIES(
    "storage_resource" = "log_s3",
    "cooldown_ttl" = "259200"
);

CREATE TABLE log_table
(
  `ts` DATETIME,
  `host` TEXT,
  `path` TEXT,
  `message` TEXT,
  INDEX idx_host (`host`) USING INVERTED,
  INDEX idx_path (`path`) USING INVERTED,
  INDEX idx_message (`message`) USING INVERTED PROPERTIES("parser" = "unicode", "support_phrase" = "true")
)
ENGINE = OLAP
DUPLICATE KEY(`ts`)
PARTITION BY RANGE(`ts`) ()
DISTRIBUTED BY RANDOM BUCKETS 60
PROPERTIES (
  "compression" = "zstd",
  "compaction_policy" = "time_series",
  "dynamic_partition.enable" = "true",
  "dynamic_partition.create_history_partition" = "true",
  "dynamic_partition.time_unit" = "DAY",
  "dynamic_partition.start" = "-30",
  "dynamic_partition.end" = "1",
  "dynamic_partition.prefix" = "p",
  "dynamic_partition.buckets" = "60",
  "dynamic_partition.replication_num" = "2", -- Not needed in compute-storage separation
  "replication_num" = "2",                    -- Not needed in compute-storage separation
  "storage_policy" = "log_policy_3day"        -- Not needed in compute-storage separation
);

3.5 ログの収集

テーブル作成後、ログの収集を開始できます。Apache Dorisは、一般的なログコレクター（Logstash、Filebeat、Kafkaなど）と統合するオープンで汎用的なStream HTTP APIを提供します。以下の表は、各収集方法の適用シナリオをまとめています：

収集方法	適用シナリオ
Logstash	豊富なフィルターとプラグインエコシステムが必要な既存のLogstashパイプライン
Filebeat	リソースに敏感なシナリオでの軽量ファイル収集
Kafka Routine Load	既にKafkaに配置されたログで、Dorisが能動的に取得するもの
カスタムプログラム（Stream Load）	社内の収集プログラムと特殊なデータソースとの統合

3.5.1 Logstashとの統合

以下の手順に従ってください：

Logstash Doris Outputプラグインをダウンロードしてインストールします。以下の方法のいずれかを選択してください：
- 直接ダウンロード：ここをクリックしてダウンロード。
- ソースからビルドし、以下のコマンドでインストール：
```
./bin/logstash-plugin install logstash-output-doris-1.2.0.gem
```
Logstashを設定します。以下の2つのファイルが必要です：
- logstash.yml：書き込みパフォーマンスを向上させるため、Logstashがログバッチングに使用するバッチサイズと遅延を設定します。
```
pipeline.batch.size: 1000000
pipeline.batch.delay: 10000
```

logstash_demo.conf: 収集するログの入力パスとApache Dorisへの出力設定を構成します。

  ```text
  input {
      file {
      path => "/path/to/your/log"
    }
  }

  output {
    doris {
      http_hosts => [ "<http://fehost1:http_port>", "<http://fehost2:http_port>", "<http://fehost3:http_port>" ]
      user => "your_username"
      password => "your_password"
      db => "your_db"
      table => "your_table"

      # doris stream load http headers
      headers => {
      "format" => "json"
      "read_json_by_line" => "true"
      "load_to_single_tablet" => "true"
      }

      # field mapping: doris fileld name => logstash field name
      # %{} to get a logstash field, [] for nested field such as [host][name] for host.name
      mapping => {
      "ts" => "%{@timestamp}"
      "host" => "%{[host][name]}"
      "path" => "%{[log][file][path]}"
      "message" => "%{message}"
      }
      log_request => true
      log_speed_interval => 10
    }
  }
  ```

Logstashを実行してログを収集し、Apache Dorisに出力します：
```
./bin/logstash -f logstash_demo.conf
```

詳細な設定については、Logstash Doris Output Pluginを参照してください。

3.5.2 Filebeatとの統合

以下の手順に従ってください：

Apache Dorisへの出力をサポートするFilebeatバイナリを取得します。こちらをクリックしてダウンロードするか、Apache Dorisソースからビルドできます。

Filebeatを設定します。メインファイルはfilebeat_demo.ymlで、収集されたログの入力パスとApache Dorisへの出力設定を構成します：

# input
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /path/to/your/log
  # multiline can join lines that span multiple lines (such as Java stack traces) into a single log
  multiline:
    type: pattern
    # Behavior: lines starting with yyyy-mm-dd HH:MM:SS are treated as a new log; others are appended to the previous log
    pattern: '^[0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}'
    negate: true
    match: after
    skip_newline: true

processors:
# Use the js script plugin to replace \t in the log with spaces, to avoid JSON parsing errors
- script:
    lang: javascript
    source: >
        function process(event) {
            var msg = event.Get("message");
            msg = msg.replace(/\t/g, "  ");
            event.Put("message", msg);
        }
# Use the dissect plugin for simple log parsing
- dissect:
    # 2024-06-08 18:26:25,481 INFO (report-thread|199) [ReportHandler.cpuReport():617] begin to handle
    tokenizer: "%{day} %{time} %{log_level} (%{thread}) [%{position}] %{content}"
    target_prefix: ""
    ignore_failure: true
    overwrite_keys: true

# queue and batch
queue.mem:
  events: 1000000
  flush.min_events: 100000
  flush.timeout: 10s

# output
output.doris:
  fenodes: [ "http://fehost1:http_port", "http://fehost2:http_port", "http://fehost3:http_port" ]
  user: "your_username"
  password: "your_password"
  database: "your_db"
  table: "your_table"
  # output string format
  ## %{[agent][hostname]} %{[log][file][path]} are metadata that come with filebeat
  ## Another commonly used filebeat metadata is the collection timestamp %{[@timestamp]}
  ## %{[day]} %{[time]} are the fields parsed by the dissect step above
  codec_format_string: '{"ts": "%{[day]} %{[time]}", "host": "%{[agent][hostname]}", "path": "%{[log][file][path]}", "message": "%{[message]}"}'
  headers:
    format: "json"
    read_json_by_line: "true"
    load_to_single_tablet: "true"

ログを収集してApache Dorisに出力するためにFilebeatを実行します：
```
chmod +x filebeat-doris-2.1.1
./filebeat-doris-2.1.1 -c filebeat_demo.yml
```

より詳細な設定については、Beats Doris Output Pluginを参照してください。

3.5.3 Kafkaとの統合

JSON形式のログをKafkaメッセージキューに書き込み、その後Kafka Routine Loadを作成してApache DorisがKafkaからデータを能動的に取得するようにします。

以下の例を参照してください。property.*はLibrdkafkaクライアントの設定です。実際のKafkaクラスタに応じて調整してください：

-- Prepare the kafka cluster and topic log__topic_
-- Create a routine load that imports data from kafka log__topic_ into the log_table table
CREATE ROUTINE LOAD load_log_kafka ON log_db.log_table
COLUMNS(ts, host, path, message)
PROPERTIES (
"max_batch_interval" = "60",
"max_batch_rows" = "20000000",
"max_batch_size" = "1073741824",
"load_to_single_tablet" = "true",
"format" = "json"
)
FROM KAFKA (
"kafka_broker_list" = "host:port",
"kafka_topic" = "log__topic_",
"property.group.id" = "your_group_id",
"property.security.protocol"="SASL_PLAINTEXT",
"property.sasl.mechanism"="GSSAPI",
"property.sasl.kerberos.service.name"="kafka",
"property.sasl.kerberos.keytab"="/path/to/xxx.keytab",
"property.sasl.kerberos.principal"="<xxx@yyy.com>"
);
-- Check the status of the routine load
SHOW ROUTINE LOAD;

Kafkaの設定詳細については、Routine Loadを参照してください。

3.5.4 カスタムプログラムを使用したログ収集

一般的なログコレクターとの統合に加えて、HTTP API Stream Loadを使用してカスタムプログラム経由でログをインポートすることもできます：

curl
--location-trusted
-u username:password
-H "format:json"
-H "read_json_by_line:true"
-H "load_to_single_tablet:true"
-H "timeout:600"
-T logfile.json
http://fe_host:fe_http_port/api/log_db/log_table/_stream_load

カスタムプログラムを使用する場合、以下の重要なポイントに注意してください：

HTTP認証にはBasic Authを使用してください。値はecho -n 'username:password' | base64コマンドで計算できます。
データ形式をJSONとして指定するために、HTTPヘッダーformat:jsonを設定してください。
1行に1つのJSONを指定するために、HTTPヘッダーread_json_by_line:trueを設定してください。
各取り込みを単一のバケットに書き込み、小さなファイルを減らすために、HTTPヘッダーload_to_single_tablet:trueを設定してください。
クライアントが100 MBから1 GBのバッチを送信することを推奨します。Apache Doris 2.1以降では、サーバー側のGroup Commit機能により、クライアント側のバッチサイズを削減できます。

3.6 ログのクエリと分析

3.6.1 ログクエリ

Apache Dorisは標準SQLをサポートしています。MySQLクライアント、JDBC、またはその他のツールを通じてクラスターに接続し、SQLクエリを実行できます：

mysql -h fe_host -P fe_mysql_port -u your_username -Dyour_db_name

参考用に5つの一般的なSQLクエリコマンドを以下に示します：

最新の10件のレコードを表示：

SELECT * FROM your_table_name ORDER BY ts DESC LIMIT 10;

hostが8.8.8.8である最新の10件のレコードをクエリする：

SELECT * FROM your_table_name WHERE host = '8.8.8.8' ORDER BY ts DESC LIMIT 10;

messageフィールドにerrorまたは404が含まれる最新の10件のレコードを検索します。MATCH_ANYはApache Dorisにおける全文検索のSQL構文で、引数内のいずれかのキーワードにマッチします：
```
SELECT * FROM your_table_name WHERE message MATCH_ANY 'error 404'
ORDER BY ts DESC LIMIT 10;
```
messageフィールドにimageとfaqの両方が含まれる最新の10件のレコードを検索します。MATCH_ALLはApache Dorisにおける全文検索のSQL構文で、引数内のすべてのキーワードにマッチします：
```
SELECT * FROM your_table_name WHERE message MATCH_ALL 'image faq'
ORDER BY ts DESC LIMIT 10;
```
messageフィールドがimageとfaqの両方を含む最新の10件のレコードを検索します。MATCH_PHRASEはApache Dorisにおける全文検索のSQL構文で、引数内のすべてのキーワードを同じ順序でマッチさせます。例えば、a image faq bはマッチしますが、a faq image bはマッチしません：
```
SELECT * FROM your_table_name WHERE message MATCH_PHRASE 'image faq'
ORDER BY ts DESC LIMIT 10;
```

3.6.2 ビジュアルログ解析

一部のサードパーティベンダーは、Apache Dorisベースのビジュアルログ解析プラットフォームを提供しており、Kibana Discoverライクなログ検索・解析インターフェースを含み、直感的で使いやすい探索的ログ解析体験を提供します：

WebUI

フルテキスト検索とSQLモードの両方をサポート
タイムピッカーとヒストグラムでログクエリの時間範囲選択をサポート
JSONやテーブルに展開可能な豊富なログ詳細ビューをサポート
ログデータのコンテキストでのインタラクティブなクリック追加・クリック削除フィルタに対応
検索結果内のフィールドのトップ値を表示し、異常の発見とさらなるドリルダウンを支援

さらなるサポートが必要な場合は、dev@doris.apache.orgまでお問い合わせください。

4. FAQ

Q1: ログシナリオにおけるApache DorisとElasticsearchのコア差分は何ですか？

A: Dorisの書き込みスループットはElasticsearchの3～5倍で、ストレージコストはElasticsearchの約20%に過ぎません。DorisはまたスタンダードSQLとMySQLプロトコルをサポートし、より強力な解析機能を提供します。ホット・コールド階層化により、コールドデータをS3/HDFSに移動し、ストレージコストをさらに削減できます。

Q2: ログフィールドが頻繁に変更されます。どのように対処すべきですか？

A: Light Schema Changeを使用して、トップレベルフィールドに対するADD/DROP COLUMNおよびADD/DROP INDEXを数秒で実行できます。動的ネストフィールドにはVARIANT型を使用し、フィールド名と型を自動検出し、VARIANTでの転置インデックスをサポートします。

Q3: バケット数はどのように選択すべきですか？

A: バケット数をクラスター内の総ディスク数の約3倍に設定し、バケットあたり約5GBの圧縮データとします。これをDISTRIBUTED BY RANDOMとシングルタブレット書き込みと組み合わせて、バッチ書き込み効率を向上させます。

Q4: ホット・コールド階層化ポリシーのcooldown_ttlの単位は何ですか？

A: 単位は秒です。例えば、259200は3日を意味し、その後データは自動的にストレージポリシーで指定されたオブジェクトストレージロケーションにコールド化されます。

Q5: 書き込み側のバッチサイズはどのように選択すべきですか？

A: リクエストあたり100MBから1GBのバッチサイズが推奨されます。Apache Doris 2.1以降では、サーバーサイドのGroup Commit機能を有効にして、クライアント側でより小さなバッチサイズを使用できます。

5. トラブルシューティング

症状	可能な原因	推奨アクション
高同時実行取り込みがトランザクション制限を超過	`max_running_txn_num_per_db`のデフォルト値が小さすぎる	`max_running_txn_num_per_db = 10000`を引き上げ
BEハートビートが頻繁にタイムアウト	重いログ書き込み圧力によりBEが一時的に応答不能	`max_backend_heartbeat_failure_tolerance_count = 10`を引き上げ
書き込みで多くの小ファイル/ランダムI/Oが発生	書き込みバッファが小さすぎるかシングルタブレット取り込みが未使用	`write_buffer_size = 1073741824`を引き上げ、`load_to_single_tablet:true`を設定
コンパクションが遅く書き込みに影響	コンパクションスレッド不足または低優先度スケジューリング制限	`max_cumu_compaction_threads`をCPUコア数/4に設定、`enable_compaction_priority_scheduling`を無効化
インデックスメモリ使用量が過大	データキャッシュとインデックスキャッシュがメモリを競合	`disable_storage_page_cache`を無効化、`inverted_index_searcher_cache_limit = 30%`を制限
フルテキスト検索のフレーズクエリが動作しない	インデックスで`support_phrase`が有効でない	インデックス作成時に`"support_phrase" = "true"`を設定
オートバケッティングでバケットが少なすぎてホットスポットが発生	`autobucket_min_buckets`が小さすぎる	`autobucket_min_buckets = 10`を引き上げ

1. 全体アーキテクチャ​

2. 機能と利点​

2.1 高性能と低コスト​

2.2 強力な分析機能​

2.3 柔軟なスキーマ​

3. 運用ガイド​

3.1 リソースの見積もり​

例: 1日100TBの新規ログのリソース見積もり​

3.2 クラスタのデプロイ​

3.3 FEとBE設定の調整​

3.3.1 FE設定の調整​

3.3.2 BE設定の調整​

3.4 テーブルの作成​

3.4.1 パーティショニングとバケット化パラメータの設定​

3.4.2 圧縮パラメータの設定​

3.4.3 コンパクションパラメータの設定​

3.4.4 インデックスの構築と設定​

3.4.5 ストレージポリシーの設定​

3.4.6 完全なテーブル作成例​

3.5 ログの収集​

3.5.1 Logstashとの統合​

3.5.2 Filebeatとの統合​

3.5.3 Kafkaとの統合​

3.5.4 カスタムプログラムを使用したログ収集​

3.6 ログのクエリと分析​

3.6.1 ログクエリ​

3.6.2 ビジュアルログ解析​

4. FAQ​

5. トラブルシューティング​