Ciao a tutti. Stiamo sviluppando un prodotto per l'analisi del traffico offline. Nel progetto c'è un compito legato all'analisi statistica dei percorsi dei visitatori nelle aree.
Nell'ambito di questo compito, gli utenti possono fare richieste al sistema del tipo:
- quanti visitatori sono passati dall'area "A" all'area "B";
- quanti visitatori sono passati dall'area "A" all'area "B" passando per l'area "C" e poi per l'area "D";
- quanto tempo ha impiegato un visitatore di un certo tipo a passare dall'area "A" all'area "B".
e una serie di richieste analitiche simili.
Il movimento del visitatore tra le aree rappresenta un grafo diretto. Dopo aver fatto alcune ricerche online, ho scoperto che i database grafici vengono utilizzati anche per report analitici. Ho avuto voglia di vedere come si comporterebbero i database grafici di fronte a tali richieste (TL;DR; male).
Ho scelto di utilizzare il database , come rappresentante eccezionale dei database grafici open-source, che si basa su una pila di tecnologie consolidate, che (secondo me) potrebbero garantirle buone prestazioni operative:
- backend del database BerkeleyDB, Apache Cassandra, Scylla;
- indici complessi possono essere memorizzati in Lucene, Elasticsearch, Solr.
Gli autori di JanusGraph affermano che è adatto sia per OLTP che per OLAP.
Ho lavorato con BerkeleyDB, Apache Cassandra, Scylla ed ES, inoltre, questi prodotti sono spesso utilizzati nei nostri sistemi, quindi ero ottimista riguardo al test di questo database grafico. Mi è sembrato insolito scegliere BerkeleyDB invece di RocksDB, ma probabilmente è legato ai requisiti di transazione. In ogni caso, per un utilizzo scalabile e produttivo viene consigliato di usare il backend su Cassandra o Scylla.
Non ho considerato Neo4j, poiché per la clustering è necessaria la versione commerciale, quindi il prodotto non è open-source.
I database grafici dicono: "Se qualcosa sembra un grafo, trattalo come un grafo!" — meraviglioso!
Inizialmente ho disegnato un grafo che seguiva proprio i canoni dei database grafici:
C'è un'entità Zone, responsabile per l'area. Se ZoneStep appartiene a questa Zone, allora la cita. Nell'entità Area, ZoneTrack, Persona non badateci, appartengono al dominio e non vengono considerate nel test. In definitiva, a una struttura grafica del genere, la richiesta di ricerca di catene apparirebbe così:
g.V().hasLabel('Zone').has('id',0).in_()
.repeat(__.out()).until(__.out().hasLabel('Zone').has('id',19)).count().next()Che in russo sarebbe più o meno così: trova Zone con ID=0, prendi tutti i vertici da cui parte un arco (ZoneStep), procedi senza tornare indietro fino a trovare ZoneStep da cui parte un arco verso Zone con ID=19, conta il numero di tali catene.
Non pretendo di conoscere tutte le sfumature della ricerca nei grafi, ma questa richiesta è stata generata sulla base di questo libro ().
Ho caricato 50.000 tracce lunghe da 3 a 20 punti nel database grafico JanusGraph, utilizzando il backend BerkeleyDB, creando indici secondo .
Script per il caricamento 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)È stata utilizzata una VM con 4 core e 16 GB di RAM su SSD. JanusGraph è stato distribuito con il seguente comando:
docker run --name janusgraph -p8182:8182 janusgraph/janusgraph:latestIn questo caso, i dati e gli indici utilizzati per la ricerca per corrispondenza esatta sono memorizzati in BerkeleyDB. Eseguendo la richiesta precedente, ho ottenuto un tempo pari a diverse decine di secondi.
Avviando 4 degli script sopra in parallelo, sono riuscito a trasformare il database in una zucca con un divertente flusso di stack trace Java (e tutti noi amiamo leggere gli stack trace Java) nei log di Docker.
Riflettendo, ho deciso di semplificare lo schema del grafo nel seguente modo:
Decidendo che la ricerca per attributi dell'entità sarebbe stata più veloce della ricerca per archi. Alla fine, la mia richiesta è diventata la seguente:
g.V().hasLabel('ZoneStep').has('id',0).repeat(__.out().simplePath()).until(__.hasLabel('ZoneStep').has('id',19)).count().next()Che in russo sarebbe più o meno così: trova ZoneStep con ID=0, procedi senza tornare indietro fino a trovare ZoneStep con ID=19, conta il numero di tali catene.
Ho anche semplificato lo script di caricamento sopra, per evitare di creare collegamenti inutili, limitandomi agli attributi.
La query stava ancora impiegando diversi secondi, il che era del tutto inaccettabile per il nostro compito, poiché per le richieste AdHoc di tipo arbitrario non andava affatto bene.
Ho provato a implementare JanusGraph usando Scylla, come la realizzazione più veloce di Cassandra, ma anche questo non ha portato a modifiche significative delle prestazioni.
Pertanto, nonostante 'sembri un grafo', non sono riuscito a far elaborare questo database grafico rapidamente. Presumo che ci sia qualcosa che non so e che JanusGraph potrebbe eseguire questa ricerca in frazioni di secondo, tuttavia, io non ci sono riuscito.
Poiché era comunque necessario risolvere il problema, ho iniziato a pensare ai JOIN e ai pivot delle tabelle, il che non era molto incoraggiante in termini di eleganza, ma potrebbe essere una soluzione praticabile nella pratica.
Nel nostro progetto utilizziamo già Apache ClickHouse, quindi ho deciso di testare le mie scoperte su questo DBMS analitico.
Ho implementato ClickHouse seguendo una semplice ricetta:
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-serverHo creato un DB e una tabella del tipo:
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 = 8192L'ho popolata con i dati usando il seguente script:
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
)Poiché le inserzioni avvengono in batch, il riempimento è stato molto più veloce rispetto a JanusGraph.
Ho costruito due query utilizzando il JOIN. Per passare dal punto A al punto B:
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.whenPer passare attraverso 3 punti:
SELECT s3.person,
s1z,
s1w,
s2z,
s2w,
s3.zone,
s3.when
FROM
(SELECT s1.person AS person,
s1.zone AS s1z,
s1.when AS s1w,
s2.zone AS s2z,
s2.when AS s2w
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 = 3)) AS s2 USING person
WHERE s1.when <= s2.when) p ANY INNER JOIN
(SELECT *
FROM steps
WHERE (area = 0)
AND (zone = 19)) AS s3 USING person
WHERE p.s2w <= s3.whenLe query, ovviamente, sembrano piuttosto spaventose, per un utilizzo reale è necessario creare un generatore di wrapper. Tuttavia, funzionano e funzionano rapidamente. Entrambe le query vengono eseguite in meno di 0.1 secondi. Ecco un esempio del tempo di esecuzione di una query per count(*) attraversando 3 punti:
SELECT count(*)
FROM
(
SELECT
s1.person AS person,
s1.zone AS s1z,
s1.when AS s1w,
s2.zone AS s2z,
s2.when AS s2w
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 = 3)
) AS s2 USING (person)
WHERE s1.when <= s2.when
) AS p
ANY INNER JOIN
(
SELECT *
FROM steps
WHERE (area = 0) AND (zone = 19)
) AS s3 USING (person)
WHERE p.s2w <= s3.when
┌─count()─┐
│ 11592 │
└─────────┘1 righe nel set. Tempo trascorso: 0.068 sec. Elaborati 250.03 mila righe, 8.00 MB (3.69 milioni di righe/s., 117.98 MB/s.)Osservazione sugli IOPS. Durante il riempimento dei dati, JanusGraph generava un numero piuttosto elevato di IOPS (1000-1300 per quattro thread di popolamento dei dati) e l'IOWAIT era piuttosto elevato. Nel frattempo, ClickHouse generava un carico minimo sul sottosistema disk.
Conclusione
Abbiamo deciso di utilizzare ClickHouse per gestire questo tipo di query. Possiamo sempre ottimizzare ulteriormente le query utilizzando viste materializzate e parallelizzazione, effettuando un pre-elaborazione del flusso degli eventi utilizzando Apache Flink prima di caricarli in ClickHouse.
Le prestazioni sono così buone che probabilmente non dovremo nemmeno pensare a pivotare le tabelle in modo programmatico. In precedenza, dovevamo fare pivot dei dati estratti da Vertica mediante esportazione in Apache Parquet.
Sfortunatamente, un altro tentativo di utilizzare un database grafico non ha avuto successo. Non ho trovato che JanusGraph avesse un ecosistema amichevole che consentisse di familiarizzare rapidamente con il prodotto. Inoltre, la configurazione del server avviene seguendo il tradizionale metodo Java, che farà piangere lacrime di sangue chi non è esperto in Java.
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 è qui per sostituire Gryo e 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 e Graphson, ultime versioni
- { 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] }}
# Versioni di serializzazione più vecchie per compatibilità:
- { 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}Sono riuscito a "installare" casualmente BerkeleyDB versione JanusGraph.
La documentazione presenta delle lacune riguardanti gli indici, poiché la gestione degli indici richiede di eseguire manovre piuttosto strane in Groovy. Ad esempio, la creazione di un indice deve avvenire scrivendo codice nella console Gremlin (che, tra l'altro, non funziona out-of-the-box). Dalla documentazione ufficiale di JanusGraph:
graph.tx().rollback() //Non creare nuovi indici mentre una transazione è attiva
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()
//Aspetta che l'indice diventi disponibile
ManagementSystem.awaitGraphIndexStatus(graph, 'byNameComposite').call()
ManagementSystem.awaitGraphIndexStatus(graph, 'byNameAndAgeComposite').call()
//Reindicizza i dati esistenti
mgmt = graph.openManagement()
mgmt.updateIndex(mgmt.getGraphIndex("byNameComposite"), SchemaAction.REINDEX).get()
mgmt.updateIndex(mgmt.getGraphIndex("byNameAndAgeComposite"), SchemaAction.REINDEX).get()
mgmt.commit()Epifania
In un certo senso, l'esperimento sopra riportato è come paragonare mele a pere. Se ci pensi, un DB a grafo esegue operazioni diverse per ottenere gli stessi risultati. Tuttavia, nel contesto dei test, ho effettuato anche un esperimento con una query del tipo:
g.V().hasLabel('ZoneStep').has('id',0)
.repeat(__.out().simplePath()).until(__.hasLabel('ZoneStep').has('id',1)).count().next()che riflette l'accessibilità passo dopo passo. Tuttavia, anche su tali dati, il DB a grafo mostrava risultati che superavano i pochi secondi... Questo, naturalmente, è legato al fatto che c'erano percorsi come 0 -> X -> Y ... -> 1, che il motore a grafo controllava anch'esso.
Anche per una query del tipo:
g.V().hasLabel('ZoneStep').has('id',0).out().has('id',1)).count().next()non sono riuscito a ottenere una risposta performante con un tempo di elaborazione inferiore a un secondo.
La morale di questa storia è che una bella idea e una modellazione paradigmatica non portano al risultato desiderato, che invece viene dimostrato con una notevole efficienza nell'esempio di ClickHouse. L'esempio presentato in questo articolo è un esplicito antipattern per i DB a grafo, anche se appare adatto per la modellazione nella loro psiche.
Fonte: habr.com
