ログストレージと分析

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

このシナリオにおいて、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圧縮アルゴリズムを採用したカラムナストレージで5〜10の圧縮率を実現し、Elasticsearchの1.5を大幅に上回ります。ホット・コールド階層機能により履歴ログを自動的にオブジェクトストレージに移動し、コールドデータのストレージコストを70%以上削減できます。全体のストレージコストはElasticsearchのわずか約20%です。
より高いクエリ性能: Dorisは全文検索パイプラインを簡素化し、ログシナリオで不要な関連性スコアリングなどのアルゴリズムをスキップします。「指定されたキーワードを含む最新の100件のログ」などの典型的なクエリに対して、Dorisはプランニングと実行レイヤーで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個のデータディスクに分散してディスクあたりの容量を計算します。

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

以下の条件を例とします：日次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クラスターをデプロイできます。物理または仮想マシンへのデプロイが推奨されます；手動デプロイ手順については、Manual deploymentを参照してください。

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 configurationを参照してください。

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 configurationを参照してください。

3.4 テーブルの作成

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

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

パーティショニング:

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

バケッティング:

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

より多くのパーティショニングとバケッティング情報については、データパーティショニングを参照してください。

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が能動的に取得する場合
Custom program (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"

Filebeatを実行してログを収集し、Apache Dorisに出力します：
```
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コマンドで計算できます。
HTTPヘッダーformat:jsonを設定して、データフォーマットをJSONとして指定します。
HTTPヘッダーread_json_by_line:trueを設定して、1行につき1つのJSONを指定します。
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 リソースの見積もり​

例: 日次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. トラブルシューティング​