Analysieren und visualisieren Sie CSV-Protokolle mit Excel Elastic Stack (Docker-Compose) - (1. Zeile: Datum, 2. und nachfolgende Zeilen: CSV-Daten) Datum wird jeder Zeile nach der 2. Zeile als Zeitstempelfeld hinzugefügt.

Einführung

Vielen Dank! Ein Ingenieur, der für den Produktinspektionsprozess in der Abteilung Produktionstechnik verantwortlich ist. Dies ist eine Fortsetzung von Analysieren und Visualisieren von CSV-Protokollen mit Excel Elastic Stack (Docker-Compose) - Was ist Elastic Stack?.

Zielgruppe

Dieser Artikel richtet sich an Personen, die Elastic Stack noch nicht kennen und darüber nachdenken, ihn auszuprobieren.

Inhalt dieses Artikels

In diesem Abschnitt wird beschrieben, wie Sie das erste Datum solcher CSV-Daten als Zeitstempel jeder Zeile behandeln.

Date,2020/10/28,20:19:18
10,Test1,130.1,OK
20,Test2,1321,OK
30,Test3,50.2,OK
End
Date,2020/10/29,10:30:50
10,Test1,140.4,OK
20,Test2,1300,OK
30,Test3,50.0,OK
End
Date,2020/10/29,11:40:50
10,Test1,141.1,OK
20,Test2,1310,OK
30,Test3,55.8,NG
End

Ich habe eine Reihe von Einstellungsdateien in GitLab eingefügt. Bitte beziehen Sie sich darauf. Klicken Sie hier für das Repository-> Elastic-Stack

Politik

Ich habe es auf zwei Arten untersucht.

• Richtlinie 1: Zeile für Zeile in den Logstash einlesen. Speichern Sie, wenn das Datum eintrifft, und stellen Sie das gespeicherte Datum in der folgenden Zeile im Feld Zeitstempel ein.
• Richtlinie 2: Verwenden Sie Filebeat, um End als ein Ereignis an logstash zu senden, und logstash, um es Zeile für Zeile zu zerlegen.

Richtlinie 1: Zeile für Zeile mit logstash lesen

Zusammenfassend hat diese Methode nicht funktioniert. aggregat Da Sie mit Filtern Informationen zwischen mehreren Ereignissen austauschen können. In einigen Fällen ist es möglich.

Im offiziellen Beispiel werden die Zahlen von Anfang bis Ende hinzugefügt und das Ergebnis des Hinzufügens zum Ereignisfeld von TASK_END festgelegt.

 INFO - 12345 - TASK_START - start
 INFO - 12345 - SQL - sqlQuery1 - 12
 INFO - 12345 - SQL - sqlQuery2 - 34
 INFO - 12345 - TASK_END - end

Sie können jedoch Daten zwischen Ereignissen austauschen, indem Sie die task_id des Aggregatfilters auf "12345" setzen. In dieser Situation war das Äquivalent task_id in den Daten nicht vorhanden und konnte nicht verwendet werden.

Richtlinie 2: Kombinieren Sie mehrere Zeilen mit Filebeat

Es ist eine erfolgreiche Methode. Mithilfe von [mehrzeilig] von filebeat (https://www.elastic.co/guide/en/beats/filebeat/current/multiline-examples.html) können mehrere Textzeilen zu einem \ n Trennzeichen kombiniert werden. Es kann ein Ereignis sein. Nachfolgend kann es durch die Regeln von drei Zeilen zusammengefasst werden.

  multiline.pattern: (End)
  multiline.negate: true
  multiline.match: before

Der Artikel von @ datake913 Zusammenfassung der mehrzeiligen Einstellungen, die mehrere Zeilen mit Filebeat verarbeiten ist eine Tabelle und leicht zu verstehen. Ich werde sie daher als "Muster" bezeichnen: Aufeinanderfolgende Zeilen, die nicht übereinstimmen (END), werden vor der nächsten übereinstimmenden Zeile hinzugefügt. " Es kann wie unten gezeigt zu einer Zeile zusammengefasst werden.

Date,2020/10/28,20:19:18\n10,Test1,130.1,OK\n20,Test2,1321,OK\n30,Test3,50.2,OK\nEnd

Fortsetzung von Richtlinie 2: Zerlegen Sie Multiline mit Logstash

Um den Zeitstempel zu analysieren, verwenden Sie De-Excel Elastic Stack (Docker-Compose), um CSV-Protokolle zu analysieren und zu visualisieren --- Analysieren Sie "Jahr / Monat / Tag, Stunde: Minute: Sekunde" in mehreren Zeilen mit Grok-Filter, japanische Zeit Behandeln als.

Wenn Sie den CSV-Filter so wie er ist auf Multiline anwenden, können Sie nur "Datum, 2020/10 / 28,20: 19: 18" bis zum ersten \ n analysieren. split Mithilfe eines Filters können Sie Multiline erneut zerlegen und in mehrere Ereignisse aufteilen. ..

Als letzter Prozess werden Zeilen, die mit einer Nicht-Nummer beginnen, mit dem Filter drop und mutate Die Typkonvertierung wird vom String-Typ mit einem Filter durchgeführt.

logstash.conf

filter {
  grok {
    patterns_dir => ["/opt/logstash/extra_patterns"]
    match => { "message" => "%{TIMESTAMP_JP:read_timestamp}" }
  }

  date {
    match => ["read_timestamp", "yyyy/MM/dd,HH:mm:ss"]
    timezone => "Asia/Tokyo"
    target => "@timestamp"
  }

  split{}

  csv {
    columns => ["Step","TestName","Value1","Judge"]
    separator => ","
  }

  if [Step] !~ /\d+/ {
    drop{}
  }

  mutate {
    convert => {
      "Step" => "integer"
      "Value1" => "float"
    }
  }
}

Schließlich

Sie können jetzt mehreren Zeilen denselben Zeitstempel mit mehreren Zeilen und Teilen zuweisen. In zukünftigen Artikeln möchte ich Gegenmaßnahmen für die folgenden Heap-Fehler einführen.

java.lang.OutOfMemoryError: Java heap space
Dumping heap to java_pid1.hprof ...
Heap dump file created [3178685347 bytes in 34.188 secs]
warning: thread "[main]>worker11" terminated with exception (report_on_exception is true):
warning: thread "[main]>worker4" terminated with exception (report_on_exception is true):
java.lang.OutOfMemoryError: Java heap space