Experiment zur Überprüfung der Anwendbarkeit der Graph-Datenbank JanusGraph zur Lösung des Problems der Suche nach passenden Wegen

Hallo zusammen. Wir entwickeln ein Produkt zur Analyse des Offline-Verkehrs. In diesem Projekt gibt es eine Aufgabe, die mit der statistischen Analyse der Bewegungswege von Besuchern in den Bereichen verbunden ist.

Im Rahmen dieser Aufgabe können die Benutzer der System Anfragen dieser Art stellen:

• Wie viele Besucher von Bereich "A" nach Bereich "B" gegangen sind;
• Wie viele Besucher von Bereich "A" über Bereich "C" und dann über Bereich "D" nach Bereich "B" gegangen sind;
• Wie viel Zeit es einen bestimmten Typ von Besucher gekostet hat, um von Bereich "A" nach Bereich "B" zu gelangen.

und eine Reihe ähnlicher analytischer Anfragen.

Die Bewegung der Besucher zwischen den Bereichen stellt einen gerichteten Graphen dar. Nachdem ich im Internet recherchiert habe, habe ich herausgefunden, dass Graph-Datenbanken auch für analytische Berichte verwendet werden. Dadurch hatte ich das Verlangen zu sehen, wie Graph-Datenbanken mit solchen Anfragen umgehen werden (TL;DR; nicht gut).

Ich habe mich für die Verwendung der Datenbank entschieden, JanusGraph, als herausragendes Beispiel einer graphbasierten Open-Source-Datenbank, die auf einem Stack aus ausgereiften Technologien beruht, die (meiner Meinung nach) ihr anständige Betriebsmerkmale bieten sollten:

• Backend-Speicher BerkeleyDB, Apache Cassandra, Scylla;
• Komplexe Indizes können in Lucene, Elasticsearch, Solr gespeichert werden.

Die Autoren von JanusGraph schreiben, dass es sowohl für OLTP als auch für OLAP geeignet ist.

Ich habe mit BerkeleyDB, Apache Cassandra, Scylla und Elasticsearch gearbeitet; darüber hinaus werden diese Produkte häufig in unseren Systemen verwendet, weshalb ich optimistisch auf das Testen dieser graphbasierten Datenbank blickte. Die Wahl von BerkeleyDB anstelle von RocksDB erschien mir merkwürdig, aber wahrscheinlich hängt das mit den Anforderungen an die Transaktionen zusammen. In jedem Fall wird für skalierbare, produktive Nutzung empfohlen, das Backend von Cassandra oder Scylla zu verwenden.

Neo4j habe ich nicht in Betracht gezogen, da für die Clusterung eine kommerzielle Version erforderlich ist, also ist das Produkt nicht offen.

Graph-Datenbanken sagen: "Wenn etwas wie ein Graph aussieht — behandeln Sie es wie einen Graphen!" — schön!

Zuerst habe ich einen Graphen gezeichnet, der genau nach den Regeln von graphbasierten Datenbanken erstellt wurde:

Es gibt eine Entität Zone, die für den Bereich verantwortlich ist. Wenn ZoneStep zu diesem gehört, verweist es auf sie. Bei den Entitäten ZoneArea ZoneTrack, Person, sollte man keine Aufmerksamkeit schenken, sie gehören zum Domain und werden im Rahmen des Tests nicht betrachtet. Insgesamt würde eine Anfrage zur Suche von Ketten in einer solchen graphischen Struktur wie folgt aussehen: Achten Sie nicht darauf, sie gehören zur Domain und werden im Rahmen des Tests nicht berücksichtigt. Insgesamt würde eine Suchanfrage nach derartigen Graphstrukturen folgendermaßen aussehen:

g.V().hasLabel('Zone').has('id',0).in_()
       .repeat(__.out()).until(__.out().hasLabel('Zone').has('id',19)).count().next()

Das klingt auf Russisch ungefähr so: Finde die Zone mit der ID=0, nimm alle Scheitelpunkte, von denen eine Kante zu ihr führt (ZoneStep), gehe ohne Rückweg, bis du solche ZoneStep findest, von denen eine Kante zur Zone mit der ID=19 führt, zähle die Anzahl solcher Ketten.

Ich beanspruche nicht, alle Feinheiten der Suche in Graphen zu kennen, aber diese Anfrage wurde basierend auf diesem Buch generiert (https://kelvinlawrence.net/book/Gremlin-Graph-Guide.html).

Ich habe 50.000 Tracks mit einer Länge von 3 bis 20 Punkten in die Graph-Datenbank JanusGraph geladen, die BerkeleyDB als Backend verwendet, und die Indizes gemäß der Anleitung.

Script zum Laden in Python:

from random import random
from time import time

from init import g, graph

if __name__ == '__main__':

    points = []
    max_zones = 19
    zcache = dict()
    for i in range(0, max_zones + 1):
        zcache[i] = g.addV('Zone').property('id', i).next()

    startZ = zcache[0]
    endZ = zcache[max_zones]

    for i in range(0, 10000):

        if not i % 100:
            print(i)

        start = g.addV('ZoneStep').property('time', int(time())).next()
        g.V(start).addE('belongs').to(startZ).iterate()

        while True:
            pt = g.addV('ZoneStep').property('time', int(time())).next()
            end_chain = random()
            if end_chain < 0.3:
                g.V(pt).addE('belongs').to(endZ).iterate()
                g.V(start).addE('goes').to(pt).iterate()
                break
            else:
                zone_id = int(random() * max_zones)
                g.V(pt).addE('belongs').to(zcache[zone_id]).iterate()
                g.V(start).addE('goes').to(pt).iterate()

            start = pt

    count = g.V().count().next()
    print(count)

Es wurde eine VM mit 4 Kernen und 16 GB RAM auf SSD verwendet. JanusGraph wurde mit folgendem Befehl bereitgestellt:

docker run --name janusgraph -p8182:8182 janusgraph/janusgraph:latest

In diesem Fall werden die Daten und Indizes, die für die Suche nach exakten Übereinstimmungen verwendet werden, in BerkeleyDB gespeichert. Bei Ausführung der oben genannten Anfrage erhielt ich eine Zeit von mehreren Dutzend Sekunden.

Nachdem ich die oben genannten 4 Skripte parallel ausgeführt hatte, gelang es mir, die Datenbank in einen Kürbis mit einem fröhlichen Strom von Java-Stacktraces (und wir alle lesen gern Java-Stacktraces) in den Docker-Logs zu verwandeln.

Nachdem ich darüber nachgedacht hatte, beschloss ich, das Graphschema wie folgt zu vereinfachen:

Ich entschied, dass die Suche nach den Attributen der Entität schneller sein würde als die Suche über die Kanten. Letztlich wurde meine Anfrage zu:

g.V().hasLabel('ZoneStep').has('id',0).repeat(__.out().simplePath()).until(__.hasLabel('ZoneStep').has('id',19)).count().next()

Das klingt auf Russisch ungefähr so: Finde ZoneStep mit ID=0, gehe ohne Rückweg, bis du ZoneStep mit ID=19 findest, zähle die Anzahl solcher Ketten.

Das oben genannte Lade-Script habe ich ebenfalls vereinfacht, um keine überflüssigen Verbindungen zu erstellen, und mich auf die Attribute beschränkt.

Die Anfrage wurde immer noch einige Sekunden lang ausgeführt, was für unsere Aufgabe völlig inakzeptabel war, da solche Ad-hoc-Anfragen völlig ungeeignet sind.

Ich habe versucht, JanusGraph mit Scylla als der schnellsten Implementierung von Cassandra zu implementieren, aber das hat auch nicht zu nennenswerten Leistungsverbesserungen geführt.

Obwohl "es wie ein Graph aussieht", ist es mir nicht gelungen, die graphbasierte Datenbank dazu zu bringen, dies schnell zu verarbeiten. Ich nehme an, dass ich etwas nicht weiß und dass man JanusGraph dazu bringen kann, diese Suche in Bruchteilen von Sekunden auszuführen, aber mir ist das nicht gelungen.

Da das Problem dennoch gelöst werden musste, begann ich über Joins und Pivot-Tabellen nachzudenken, was aus ästhetischer Sicht nicht gerade optimistisch stimmte, aber in der Praxis durchaus funktionieren könnte.

In unserem Projekt verwenden wir bereits Apache ClickHouse, also wollte ich meine Erkenntnisse an dieser analytischen Datenbank prüfen.

Ich habe ClickHouse nach einem einfachen Rezept installiert:

sudo docker run -d --name clickhouse_1 
     --ulimit nofile=262144:262144 
     -v /opt/clickhouse/log:/var/log/clickhouse-server 
     -v /opt/clickhouse/data:/var/lib/clickhouse 
     yandex/clickhouse-server

Ich habe darin eine Datenbank und eine Tabelle vom Typ erstellt:

CREATE TABLE 
db.steps (`area` Int64, `when` DateTime64(1, 'Europe/Moscow') DEFAULT now64(), `zone` Int64, `person` Int64) 
ENGINE = MergeTree() ORDER BY (area, zone, person) SETTINGS index_granularity = 8192

Ich habe sie mit den folgenden Daten gefüllt:

from time import time

from clickhouse_driver import Client
from random import random

client = Client('vm-12c2c34c-df68-4a98-b1e5-a4d1cef1acff.domain',
                database='db',
                password='secret')

max = 20

for r in range(0, 100000):

    if r % 1000 == 0:
        print("CNT: {}, TS: {}".format(r, time()))

    data = [{
            'area': 0,
            'zone': 0,
            'person': r
        }]

    while True:
        if random() < 0.3:
            break

        data.append({
                'area': 0,
                'zone': int(random() * (max - 2)) + 1,
                'person': r
            })

    data.append({
            'area': 0,
            'zone': max - 1,
            'person': r
        })

    client.execute(
        'INSERT INTO steps (area, zone, person) VALUES',
        data
    )

Da die Einfügungen in Batches erfolgen, war das Ausfüllen viel schneller als bei JanusGraph.

Ich habe zwei Abfragen mit JOIN konstruiert. Um von Punkt A nach Punkt B zu gelangen:

SELECT s1.person AS person,
       s1.zone,
       s1.when,
       s2.zone,
       s2.when
FROM
  (SELECT *
   FROM steps
   WHERE (area = 0)
     AND (zone = 0)) AS s1 ANY INNER JOIN
  (SELECT *
   FROM steps AS s2
   WHERE (area = 0)
     AND (zone = 19)) AS s2 USING person
WHERE s1.when <= s2.when

Um über 3 Punkte zu gelangen:

WÄHLEN s3.person,
       s1z,
       s1w,
       s2z,
       s2w,
       s3.zone,
       s3.wann
VON
  (WÄHLEN s1.person AS person,
          s1.zone AS s1z,
          s1.wann AS s1w,
          s2.zone AS s2z,
          s2.wann AS s2w
   VON
     (WÄHLEN *
      VON schritte
      WO (bereich = 0)
        UND (zone = 0)) AS s1 IRGENDEIN INNER JOIN
     (WÄHLEN *
      VON schritte AS s2
      WO (bereich = 0)
        UND (zone = 3)) AS s2 VERWENDEN person
   WO s1.wann <= s2.wann) p IRGENDEIN INNER JOIN
  (WÄHLEN *
   VON schritte
   WO (bereich = 0)
     UND (zone = 19)) AS s3 VERWENDEN person
WO p.s2w <= s3.wann

Die Abfragen sehen natürlich ziemlich furchterregend aus, für eine tatsächliche Nutzung ist es erforderlich, eine Programmierung-Wrapper-Generator zu erstellen. Sie funktionieren jedoch und sind schnell. Sowohl die erste als auch die zweite Abfrage werden in weniger als 0,1 Sekunden ausgeführt. Hier ist ein Beispiel für die Ausführungszeit der Abfrage für count(*) über 3 Punkte:

WÄHLEN count(*)
VON 
(
    WÄHLEN 
        s1.person AS person, 
        s1.zone AS s1z, 
        s1.wann AS s1w, 
        s2.zone AS s2z, 
        s2.wann AS s2w
    VON 
    (
        WÄHLEN *
        VON schritte
        WO (bereich = 0) UND (zone = 0)
    ) AS s1
    IRGENDEIN INNER JOIN 
    (
        WÄHLEN *
        VON schritte AS s2
        WO (bereich = 0) UND (zone = 3)
    ) AS s2 VERWENDEN (person)
    WO s1.wann <= s2.wann
) AS p
IRGENDEIN INNER JOIN 
(
    WÄHLEN *
    VON schritte
    WO (bereich = 0) UND (zone = 19)
) AS s3 VERWENDEN (person)
WO p.s2w <= s3.wann

┌─count()─┐
│   11592 │
└─────────┘

1 Zeile im Set. Verstrichene Zeit: 0.068 Sek. Verarbeitet 250,03 Tausend Zeilen, 8.00 MB (3,69 Millionen Zeilen/s., 117,98 MB/s.)

Anmerkung zu IOPS. Beim Befüllen der Daten generierte JanusGraph eine recht hohe Anzahl an IOPS (1000-1300 für vier Datenfüllströme), und die IOWAIT war recht hoch. Gleichzeitig erzeugte ClickHouse eine minimale Belastung des Datenspeichersystems.

Fazit

Wir haben uns entschieden, ClickHouse zur Bedienung von Abfragen dieser Art zu nutzen. Wir können die Abfragen immer weiter optimieren, indem wir materialisierte Sichten und Parallelisierung verwenden, indem wir den Datenstrom mit Apache Flink vor der Ladung in ClickHouse vorverarbeiten.

Die Leistung ist so gut, dass wir wahrscheinlich sogar nicht mehr über die Pivotierung von Tabellen mit Programmiertools nachdenken müssen. Zuvor mussten wir die Daten, die aus Vertica extrahiert wurden, durch den Export in Apache Parquet pivotieren.

Leider war der nächste Versuch, eine graphenbasierte DBMS zu nutzen, nicht erfolgreich. Ich fand nicht, dass JanusGraph eine benutzerfreundliche Umgebung bietet, die es ermöglicht, sich schnell mit dem Produkt vertraut zu machen. Gleichzeitig wird für die Konfiguration des Servers die traditionelle Java-Way verwendet, was Menschen ohne Java-Kenntnisse in Tränen ausbrechen lassen wird.

host: 0.0.0.0
port: 8182
threadPoolWorker: 1
gremlinPool: 8
scriptEvaluationTimeout: 30000
channelizer: org.janusgraph.channelizers.JanusGraphWsAndHttpChannelizer

graphManager: org.janusgraph.graphdb.management.JanusGraphManager
graphs: {
  ConfigurationManagementGraph: conf/janusgraph-cql-configurationgraph.properties,
  airlines: conf/airlines.properties
}

scriptEngines: {
  gremlin-groovy: {
    plugins: { org.janusgraph.graphdb.tinkerpop.plugin.JanusGraphGremlinPlugin: {},
               org.apache.tinkerpop.gremlin.server.jsr223.GremlinServerGremlinPlugin: {},
               org.apache.tinkerpop.gremlin.tinkergraph.jsr223.TinkerGraphGremlinPlugin: {},
               org.apache.tinkerpop.gremlin.jsr223.ImportGremlinPlugin: {classImports: [java.lang.Math], methodImports: [java.lang.Math#*]},
               org.apache.tinkerpop.gremlin.jsr223.ScriptFileGremlinPlugin: {files: [scripts/airline-sample.groovy]}}}}

serializers:
# GraphBinary ersetzt Gryo und Graphson
  - { className: org.apache.tinkerpop.gremlin.driver.ser.GraphBinaryMessageSerializerV1, config: { ioRegistries: [org.janusgraph.graphdb.tinkerpop.JanusGraphIoRegistry] }}
  - { className: org.apache.tinkerpop.gremlin.driver.ser.GraphBinaryMessageSerializerV1, config: { serializeResultToString: true }}
  # Gryo und Graphson, neueste Versionen
  - { className: org.apache.tinkerpop.gremlin.driver.ser.GryoMessageSerializerV3d0, config: { ioRegistries: [org.janusgraph.graphdb.tinkerpop.JanusGraphIoRegistry] }}
  - { className: org.apache.tinkerpop.gremlin.driver.ser.GryoMessageSerializerV3d0, config: { serializeResultToString: true }}
  - { className: org.apache.tinkerpop.gremlin.driver.ser.GraphSONMessageSerializerV3d0, config: { ioRegistries: [org.janusgraph.graphdb.tinkerpop.JanusGraphIoRegistry] }}
  # Ältere Serialisierungs-Versionen für Abwärtskompatibilität:
  - { className: org.apache.tinkerpop.gremlin.driver.ser.GryoMessageSerializerV1d0, config: { ioRegistries: [org.janusgraph.graphdb.tinkerpop.JanusGraphIoRegistry] }}
  - { className: org.apache.tinkerpop.gremlin.driver.ser.GryoMessageSerializerV1d0, config: { serializeResultToString: true }}
  - { className: org.apache.tinkerpop.gremlin.driver.ser.GryoLiteMessageSerializerV1d0, config: {ioRegistries: [org.janusgraph.graphdb.tinkerpop.JanusGraphIoRegistry] }}
  - { className: org.apache.tinkerpop.gremlin.driver.ser.GraphSONMessageSerializerGremlinV2d0, config: { ioRegistries: [org.janusgraph.graphdb.tinkerpop.JanusGraphIoRegistry] }}
  - { className: org.apache.tinkerpop.gremlin.driver.ser.GraphSONMessageSerializerGremlinV1d0, config: { ioRegistries: [org.janusgraph.graphdb.tinkerpop.JanusGraphIoRegistryV1d0] }}
  - { className: org.apache.tinkerpop.gremlin.driver.ser.GraphSONMessageSerializerV1d0, config: { ioRegistries: [org.janusgraph.graphdb.tinkerpop.JanusGraphIoRegistryV1d0] }}

processors:
  - { className: org.apache.tinkerpop.gremlin.server.op.session.SessionOpProcessor, config: { sessionTimeout: 28800000 }}
  - { className: org.apache.tinkerpop.gremlin.server.op.traversal.TraversalOpProcessor, config: { cacheExpirationTime: 600000, cacheMaxSize: 1000 }}

metrics: {
  consoleReporter: {enabled: false, interval: 180000},
  csvReporter: {enabled: false, interval: 180000, fileName: /tmp/gremlin-server-metrics.csv},
  jmxReporter: {enabled: false},
  slf4jReporter: {enabled: true, interval: 180000},
  gangliaReporter: {enabled: false, interval: 180000, addressingMode: MULTICAST},
  graphiteReporter: {enabled: false, interval: 180000}}
threadPoolBoss: 1
maxInitialLineLength: 4096
maxHeaderSize: 8192
maxChunkSize: 8192
maxContentLength: 65536
maxAccumulationBufferComponents: 1024
resultIterationBatchSize: 64
writeBufferHighWaterMark: 32768
writeBufferHighWaterMark: 65536
ssl: {
  enabled: false}

Ich habe es zufällig geschafft, BerkeleyDB Version JanusGraph "zulegen".

Die Dokumentation ist ziemlich unklar in Bezug auf die Indizes, da beim Management von Indizes recht seltsame Dinge in Groovy benötigt werden. Zum Beispiel muss die Erstellung eines Index durch das Schreiben von Code in der Gremlin-Konsole erfolgen (die übrigens nicht von Anfang an funktioniert). Aus der offiziellen Dokumentation von JanusGraph:

graph.tx().rollback() 
// Erstellen Sie niemals neue Indizes, während eine Transaktion aktiv ist
mgmt = graph.openManagement()
name = mgmt.getPropertyKey('name')
age = mgmt.getPropertyKey('age')
mgmt.buildIndex('byNameComposite', Vertex.class).addKey(name).buildCompositeIndex()
mgmt.buildIndex('byNameAndAgeComposite', Vertex.class).addKey(name).addKey(age).buildCompositeIndex()
mgmt.commit()

// Warten Sie, bis der Index verfügbar ist
ManagementSystem.awaitGraphIndexStatus(graph, 'byNameComposite').call()
ManagementSystem.awaitGraphIndexStatus(graph, 'byNameAndAgeComposite').call()
// Reindexieren Sie die vorhandenen Daten
mgmt = graph.openManagement()
mgmt.updateIndex(mgmt.getGraphIndex("byNameComposite"), SchemaAction.REINDEX).get()
mgmt.updateIndex(mgmt.getGraphIndex("byNameAndAgeComposite"), SchemaAction.REINDEX).get()
mgmt.commit()

Nachwort

In gewisser Hinsicht ist das obige Experiment ein Vergleich von Äpfeln mit Birnen. Wenn man darüber nachdenkt, führt eine Graph-Datenbank andere Operationen aus, um die gleichen Ergebnisse zu erzielen. Dennoch habe ich im Rahmen der Tests auch ein Experiment mit einer Abfrage durchgeführt, die lautet:

g.V().hasLabel('ZoneStep').has('id',0)
    .repeat(__.out().simplePath()).until(__.hasLabel('ZoneStep').has('id',1)).count().next()

, die die schrittweise Erreichbarkeit widerspiegelt. Doch selbst mit solchen Daten zeigte die Graph-Datenbank Ergebnisse, die über mehrere Sekunden hinausgingen... Dies hängt sicherlich damit zusammen, dass es Pfade gab wie 0 -> X -> Y ... -> 1, die der Graph-Engine ebenfalls überprüft wurden.

Selbst für eine Anfrage wie:

g.V().hasLabel('ZoneStep').has('id',0).out().has('id',1)).count().next()

ist es mir nicht gelungen, eine reaktionsschnelle Antwort mit einer Verarbeitungszeit von weniger als einer Sekunde zu erhalten.

Die Moral der Geschichte ist, dass eine schöne Idee und paradigmatische Modellierung nicht zu den gewünschten Ergebnissen führen, die mit deutlich höherer Effizienz am Beispiel von ClickHouse demonstriert werden. Der in diesem Artikel beschriebene Anwendungsfall ist ein offensichtliches Antimuster für Graph-Datenbanken, obwohl er auf den ersten Blick als passend für die Modellierung in ihrer Paradigmen erscheint.

Quelle: habr.com