log_reduce_fl()

  • [アーティクル]

関数 log_reduce_fl() は、ログ行などの半構造化テキスト列で一般的なパターンを検索し、抽出されたパターンに従って行をクラスター化します。 検出されたパターンを含むサマリー テーブルを出力し、それぞれの頻度で上に並べ替えます。

前提条件

  • データベースで Python プラグイン を有効にする必要があります。 これは、 関数で使用されるインライン Python に必要です。

構文

T|invokelog_reduce_fl(reduce_col [,use_logram [,use_drain[,custom_regexes[,,custom_regexes_policy区切り記号[similarity_th[,,tree_depth[trigram_th[,,bigram_th ]]]]]]]]])

構文規則について詳しく知る。

パラメーター

次のパラメーターの説明は概要です。 詳細については、「アルゴリズムの 詳細 」セクションを参照してください。

名前 型 必須 説明
reduce_col string ✔️ 関数が適用される文字列列の名前。
use_logram bool Logram アルゴリズムを有効または無効にします。 既定値は true です。
use_drain bool ドレイン アルゴリズムを有効または無効にします。 既定値は true です。
custom_regexes dynamic 各入力行で検索され、それぞれの一致するシンボルに置き換えられる正規表現と置換シンボルのペアを含む動的配列。 既定値は dynamic([]) です。 既定の正規表現テーブルは、数値、IP、GUID を置き換えます。
custom_regexes_policy string 'prepend'、'append' または 'replace' のいずれか。 custom_regexesが既定の値の前に追加または置換されるかどうかを制御します。 既定値は 'prepend' です。
delimiters dynamic 区切り記号文字列を含む動的配列。 既定値は で dynamic([" "])、スペースを 1 文字の区切り記号として定義します。
similarity_th real ドレイン アルゴリズムで使用される類似性しきい値。 similarity_thを増やすと、クラスターがより洗練されます。 既定値は 0.5 です。 Drain が無効になっている場合、このパラメーターは無効になります。
tree_depth int tree_depthを増やすと、ドレイン アルゴリズムの実行時間が向上しますが、精度が低下する可能性があります。 既定値は 4 です。 Drain が無効になっている場合、このパラメーターは無効になります。
trigram_th int trigram_thを減らすと、Logram がトークンをワイルドカードに置き換える可能性が高くなります。 既定値は 10 です。 Logram が無効になっている場合、このパラメーターは無効になります。
bigram_th int bigram_thを減らすと、トークンをワイルドカードに置き換える Logram の可能性が高くなります。 既定値は 15 です。 Logram が無効になっている場合、このパラメーターは無効になります。

アルゴリズムの詳細

関数は、一般的なパターンに減らすために、複数の行を渡して実行します。 パスの一覧を次に示します。

  • 正規表現の置換: このパスでは、各行は一連の正規表現に個別に一致し、一致した各式は置換記号に置き換えられます。 既定の正規表現は、IP、数値、GUID を /<IP>、GUID>、</<NUM> に置き換えます。 ユーザーは、その前に正規表現を追加したり、新しい正規表現や空のリストに置き換えたりするには 、custom_regexes と custom_regexes_policyを変更します。 たとえば、整数 <を WNUM> set custom_regexes=pack_array('/^\d+$/', '<WNUM>')に置き換えるには、正規表現の置換セット custom_regexes_policy='replace'' を取り消します。 各行について、関数は元の式 (置換前) の一覧をジェネリック置換トークンのパラメーターとして出力します。

  • トークン化: 前の手順と同様に、各行は個別に処理され、 区切り記号のセットに基づいてトークンに分割されます。 たとえば、コンマ、ピリオド、またはセミコロンでトークンへの改行を定義するには、 delimiters=pack_array(',', '.', ';') を設定します。

  • Logram アルゴリズムの適用: このパスは省略可能で、保留中のuse_logramは true です。 大規模なスケールが必要な場合や、ログ エントリの最初のトークンにパラメーターを含めることができる場合は、Logram を使用することをお勧めします。 OTOH では、ログ エントリが短い場合は無効にします。このような場合、アルゴリズムはトークンをワイルドカードに置き換える傾向があるためです。 Logram アルゴリズムでは、3 タプルと 2 タプルのトークンが考慮されます。 3 タプルのトークンがログ行で一般的である場合 ( trigram_th 回以上表示されます)、3 つのトークンすべてがパターンの一部である可能性があります。 3 タプルがまれな場合は、ワイルドカードで置き換える必要がある変数が含まれている可能性があります。 まれな 3 タプルの場合、3 タプルに含まれる 2 タプルが出現する頻度を考慮します。 2 タプルが一般的な場合 ( bigram_th 回以上表示されます)、残りのトークンはパラメーターであり、パターンの一部ではない可能性があります。
    Logram アルゴリズムは簡単に並列化できます。 ログ コーパスには 2 つのパスが必要です。1 つ目は各 3 タプルと 2 タプルの頻度をカウントし、2 つ目は前に説明したロジックを各エントリに適用します。 アルゴリズムを並列化するには、ログ エントリをパーティション分割し、さまざまなワーカーの頻度数を統一するだけで済みます。

  • ドレイン アルゴリズムの適用: このパスは省略可能で、保留中のuse_drainは true です。 ドレインは、切り捨てられた深度プレフィックス ツリーに基づくログ解析アルゴリズムです。 ログ メッセージは長さに応じて分割され、長さごとにログ メッセージの最初の tree_depth トークンを使用してプレフィックス ツリーが作成されます。 プレフィックス トークンに一致するものが見つからなかった場合は、新しいブランチが作成されます。 プレフィックスの一致が見つかった場合は、ツリー リーフに含まれるパターンの中で最も類似したパターンを検索します。 パターンの類似性は、すべてのトークンから一致した非ワイルドカード トークンの比率によって測定されます。 最も類似したパターンの類似性が類似性しきい値 (パラメーター similarity_th) を超えている場合、ログ エントリはパターンと一致します。 このパターンの場合、関数は一致しないすべてのトークンをワイルドカードで置き換えます。 最も類似したパターンの類似性が類似性しきい値を下回る場合は、ログ エントリを含む新しいパターンが作成されます。
    さまざまなログのテストに基づいて、既定の tree_depth を 4 に設定します。 この深さを増やすと、ランタイムが向上しますが、パターンの精度が低下する可能性があります。各ノードはより多くの類似性テストを実行するため、精度は高くなりますが、速度は低下します。
    通常、Drain は効率的にパターンを一般化して削減します (ただし、並列化するのは困難です)。 ただし、プレフィックス ツリーに依存しているため、最初のトークンにパラメーターを含むログ エントリでは最適でない場合があります。 これは、ほとんどの場合、Logram を最初に適用することで解決できます。

関数の定義

関数を定義するには、次のように、コードをクエリ定義関数として埋め込むか、データベースに格納されている関数として作成します。

次の let ステートメントを使用して関数を定義します。 権限は必要ありません。

重要

let ステートメントは単独では実行できません。 その後に 表形式の式ステートメントを指定する必要があります。 の動作例 log_reduce_fl()を実行するには、「 例」を参照してください。

let log_reduce_fl=(tbl:(*), reduce_col:string,
              use_logram:bool=True, use_drain:bool=True, custom_regexes: dynamic = dynamic([]), custom_regexes_policy: string = 'prepend',
              delimiters:dynamic = dynamic(' '), similarity_th:double=0.5, tree_depth:int = 4, trigram_th:int=10, bigram_th:int=15)
{
    let default_regex_table = pack_array('(/|)([0-9]+\\.){3}[0-9]+(:[0-9]+|)(:|)', '<IP>', 
                                         '([0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12})', '<GUID>', 
                                         '(?<=[^A-Za-z0-9])(\\-?\\+?\\d+)(?=[^A-Za-z0-9])|[0-9]+$', '<NUM>');
    let kwargs = bag_pack('reduced_column', reduce_col, 'delimiters', delimiters,'output_column', 'LogReduce', 'parameters_column', '', 
                          'trigram_th', trigram_th, 'bigram_th', bigram_th, 'default_regexes', default_regex_table, 
                          'custom_regexes', custom_regexes, 'custom_regexes_policy', custom_regexes_policy, 'tree_depth', tree_depth, 'similarity_th', similarity_th, 
                          'use_drain', use_drain, 'use_logram', use_logram, 'save_regex_tuples_in_output', True, 'regex_tuples_column', 'RegexesColumn', 
                          'output_type', 'summary');
    let code = ```if 1:
        from log_cluster import log_reduce
        result = log_reduce.log_reduce(df, kargs)
    ```;
    tbl
    | extend LogReduce=''
    | evaluate python(typeof(Count:int, LogReduce:string, example:string), code, kwargs)
};
// Write your query to use the function here.

例

次の例では、 invoke 演算子 を使用して関数を実行します。 この例では、 Apache Hadoop 分散ファイル システム ログを使用します。

クエリ定義関数を使用するには、埋め込み関数定義の後で呼び出します。

let log_reduce_fl=(tbl:(*), reduce_col:string,
              use_logram:bool=True, use_drain:bool=True, custom_regexes: dynamic = dynamic([]), custom_regexes_policy: string = 'prepend',
              delimiters:dynamic = dynamic(' '), similarity_th:double=0.5, tree_depth:int = 4, trigram_th:int=10, bigram_th:int=15)
{
    let default_regex_table = pack_array('(/|)([0-9]+\\.){3}[0-9]+(:[0-9]+|)(:|)', '\<IP>', 
                                         '([0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12})', '<GUID>', 
                                         '(?<=[^A-Za-z0-9])(\\-?\\+?\\d+)(?=[^A-Za-z0-9])|[0-9]+$', '\<NUM>');
    let kwargs = bag_pack('reduced_column', reduce_col, 'delimiters', delimiters,'output_column', 'LogReduce', 'parameters_column', '', 
                          'trigram_th', trigram_th, 'bigram_th', bigram_th, 'default_regexes', default_regex_table, 
                          'custom_regexes', custom_regexes, 'custom_regexes_policy', custom_regexes_policy, 'tree_depth', tree_depth, 'similarity_th', similarity_th, 
                          'use_drain', use_drain, 'use_logram', use_logram, 'save_regex_tuples_in_output', True, 'regex_tuples_column', 'RegexesColumn', 
                          'output_type', 'summary');
    let code = ```if 1:
        from log_cluster import log_reduce
        result = log_reduce.log_reduce(df, kargs)
    ```;
    tbl
    | extend LogReduce=''
    | evaluate python(typeof(Count:int, LogReduce:string, example:string), code, kwargs)
};
//
// Finding common patterns in HDFS logs, a commonly used benchmark for log parsing
//
HDFS_log
| take 100000
| invoke log_reduce_fl(reduce_col="data")

出力

Count LogReduce 例
55356 081110 <NUM INFO DFS の値><を入力>します。FSNamesystem: BLOCK* NameSystem.delete: ip< 081110 220623 26 INFO dfs の <> invalidSet にblk_NUM> が追加されます。FSNamesystem: BLOCK* NameSystem.delete: blk_1239016582509138045が invalidSet of 10.251.123.195:50010 に追加されます
10278 081110 <NUM INFO DFS の値><を入力>します。FSNamesystem: BLOCK* NameSystem.addStoredBlock: blockMap updated: IP> is added to <blk_<NUM> size <NUM> 081110 215858 27 INFO dfs.FSNamesystem: BLOCK* NameSystem.addStoredBlock: blockMap updated: 10.250.11.85:50010 がblk_5080254298708411681 サイズに追加67108864
10256 081110 <NUM INFO DFS の値><を入力>します。DataNode$PacketResponder: ブロック blk_<NUM> 終端 081110 215858 15496 INFO dfs の PacketResponder <NUM>。DataNode$PacketResponder: Block blk_-7746692545918257727 終了用 PacketResponder 2
10256 081110 <NUM INFO DFS の値><を入力>します。DataNode$PacketResponder: IP 081110 215858< 15485 INFO dfs から <NUM> サイズ<のブロック blk_NUM> を受信しました。>DataNode$PacketResponder: /10.251.43.21 から67108864サイズの受信ブロック blk_5080254298708411681
9140 081110 <NUM INFO DFS の値><を入力> します。DataNode$DataXceiver: 受信ブロック blk_<NUM> src: <IP> dest: <IP> 081110 215858 15494 INFO dfs。DataNode$DataXceiver: 受信ブロック blk_-7037346755429293022 src: /10.251.43.21:45933 dest: /10.251.43.21:50010
3047 081110 <NUM INFO DFS の値><を入力>します。FSNamesystem: BLOCK* NameSystem.allocateBlock: /user/root/rand3/temporary/task<NUM><>m<NUM>_<NUM/part-NUM><>。 <> 081110 215858 26 INFO dfs。FSNamesystem: BLOCK NameSystem.allocateBlock: /user/root/rand3/_temporary/task_200811101024_0005_m_001805_0/part-01805. blk-7037346755429293022
1402 081110 <NUM><INFO><>: <> ブロック blk_<NUM> <><> 081110 215957 15556 INFO dfs.DataNode$DataTransfer: 10.250.15.198:50010:送信ブロック blk_-3782569120714539446 から /10.251.203.129:50010
177 081110 <NUM><INFO><>: <>><<> <*> 081110 215859 13 INFO dfs.DataBlockScanner: blk_-7244926816084627474の検証に成功しました
36 081110 <NUM><INFO><>: <><>> <ブロック <*> 081110 215924 15636 INFO dfs の場合。DataNode$BlockReceiver: ブロック blk_3991288654265301939の空のパケットを受信しています
12 081110 <NUM INFO DFS の値><を入力>します。FSNamesystem: BLOCK* <>><> <> <> <><><> <081110 215953 19 INFO dfs。FSNamesystem: BLOCK* ask 10.250.15.198:50010 blk_-3782569120714539446を datanode(s) 10.251.203.129:50010 にレプリケートする
12 081110 <NUM><INFO><>: <><> <>><> <ブロック blk_<NUM><> <> 081110 215955 18 INFO dfs.DataNode: 10.250.15.198:50010 ブロック blk_-3782569120714539446 を 10.251.203.129:50010 に転送するためのスレッドの開始
12 081110 <NUM INFO DFS の値><を入力> します。DataNode$DataXceiver: Received block blk_<NUM> src: <IP> dest: <IP> of size <NUM> 081110 215957 15226 INFO dfs.DataNode$DataXceiver: Received block blk_-3782569120714539446 src: /10.250.15.198:51013 dest: /10.250.15.198:50010 のサイズ 14474705
6 081110 <NUM><NUM><> dfs。FSNamesystem: BLOCK NameSystem.addStoredBlock: <>> <<><> <<> <> <>>SIZE <NUM> 081110 215924 27 WARN dfs.FSNamesystem: BLOCK* NameSystem.addStoredBlock: 10.251.202.134:50010 サイズのblk_2522553781740514003に対して受信した冗長な addStoredBlock 要求67108864
6 081110 <NUM INFO DFS の値><を入力>します。DataNode$DataXceiver: <>> <<> <<>>: <><><> <>> <081110 215936 15714 INFO dfs。DataNode$DataXceiver: writeBlock blk_720939897861061328受け取った例外 java.io.IOException: ストリームから読み取れませんでした
3 081110 <NUM INFO DFS の値><を入力>します。FSNamesystem: BLOCK* NameSystem.addStoredBlock: <>><> <<> <<> <>> size <NUM><>> <> <><<<>> <><>. 081110 220635 28 INFO dfs.FSNamesystem: BLOCK NameSystem.addStoredBlock: 53457811 10.250.17.177:50010 サイズの blk_-81196479666306310 に対して受信した addStoredBlock 要求が、どのファイルにも属していません。
1 081110 <NUM><NUM><> <>: <>. <>> <>>> <<<> <><> <><> <>< 081110 220631 19 WARN dfs。FSDataset: ブロック blk_-2012154052725261337を削除しようとすると予期しないエラーが発生しました。 volumeMap に BlockInfo が見つかりません。

この機能はサポートされていません。