0
我想知道是否可以使用Neo4j分析網絡流量日誌文件。 因此，我使用「sh」-library實時「Broiling」來自Bro IDS的3個不同日誌文件，並將日誌記錄導入neo4j，這似乎是使用py2neo很慢的。 CSV導入在這裏不起作用，因爲它是實時的。Neo4j - 實時導入日誌文件的寫入速度太慢？
一個示例：我正在使用tcpreplay分析一個小時的數據包捕獲文件，該文件具有幾乎4.000.000個連接。我甚至用速度的一半來演奏它。因此，在2小時後，我有大約4000000條日誌條目。現在，在分析開始後的3,5小時，我剛剛導入了由5個節點和4個關係組成的289691個圖。總而言之，大約15％的數據幾乎是時間的兩倍。
我使用py2neo和代碼看起來像以下（這是圖的一個）：
def create_conn_graph(connlog):
[...]
## Start Session
graph = Graph(bolt=True, password="neo4j")
tx = graph.begin()
############
## Nodes ##
############
## Connection Node
conn = Node("Connection", uid=connlog['uid'],
ts=connlog['ts'],
date=evt_date,
time=evt_time,
[...])
conn_properties = dict(conn)
for key in conn_properties.keys():
if conn[key] == "-" or conn[key] == "(empty)":
conn[key] = "0"
conn.update()
tx.merge(conn, "Connection", "uid")
## IP Nodes
orig = Node("IP", ip=connlog['orig_h'])
tx.merge(orig)
resp = Node("IP", ip=connlog['resp_h'])
tx.merge(resp)
## History Node
if connlog['history']:
hist_flow = history_flow(connlog['history'])
history_node = Node("History", history=connlog['history'], flow=hist_flow)
tx.merge(history_node, "History", "history")
## (Connection)-[HAS_HISTORY]->(History)
conn_hist = Relationship(conn, "HAS_HISTORY", history_node)
tx.merge(conn_hist)
## Conn_State
conn_state = Node("Conn_State", state=connlog['conn_state'], meaning=CONN_STATE[connlog['conn_state']])
tx.merge(conn_state, "Conn_State", "conn_state")
tx.commit()
tx = graph.begin()
#####################
## Relationships ##
#####################
## (IP)-[STARTS_CONNECTION]->(Connection)
orig_conn = Relationship(orig, "STARTS_CONNECTION", conn, port=connlog['orig_p'])
tx.merge(orig_conn)
## (Connection)-[CONNECTS_TO]->(IP)
conn_resp = Relationship(conn, "CONNECTS_TO", resp, port=connlog['resp_p'])
tx.merge(conn_resp)
## (Connection)-[HAS_CONN_STATE]->(Conn_State)
conn_connstate = Relationship(conn, "HAS_CONN_STATE", conn_state)
tx.merge(conn_connstate)
tx.commit()
## (Connection)-[PRODUCED]-> (DNS|HTTP)
if connlog['service'] == "dns":
graph.run("MATCH (c:Connection {uid:{uid}}), (d:DNS {uid:{uid}}) \
MERGE (c)-[:PRODUCED]->(d)",
{"uid": connlog['uid']})
if connlog['service'] == "http":
graph.run("MATCH (c:Connection {uid:{uid}}), (d:HTTP {uid:{uid}}) \
MERGE (c)-[:PRODUCED]->(d)",
{"uid": connlog['uid']})
return True
## End of create_conn_graph ########################################
if __name__ == "__main__":
logentry = {}
logfield = CONNLOG
logline = []
for line in tail("-F", LOG_DIR, _iter=True, _bg=True):
entry = line.strip().split("\t")
if line.startswith('#'):
continue
for i in range(len(logfield)):
logentry[logfield[i]] = entry[i]
create_conn_graph(logentry)
我有以下的約束和索引：
graph.run("CREATE CONSTRAINT ON (c:Connection) ASSERT c.uid IS UNIQUE")
graph.run("CREATE CONSTRAINT ON (i:IP) ASSERT i.ip IS UNIQUE")
graph.run("CREATE CONSTRAINT ON (c:Conn_State) ASSERT c.conn_state IS UNIQUE")
graph.run("CREATE CONSTRAINT ON (h:History) ASSERT h.history IS UNIQUE")
graph.run("CREATE CONSTRAINT ON (host:Host) ASSERT host.host is UNIQUE")
graph.run("CREATE CONSTRAINT ON (q:QueryType) ASSERT q.type is UNIQUE")
graph.run("CREATE CONSTRAINT ON (qc:QueryClass) ASSERT qc.class is UNIQUE")
graph.run("CREATE CONSTRAINT ON (rc:ResponseCode) ASSERT rc.code is UNIQUE")
graph.run("CREATE CONSTRAINT ON (ic:InfoCode) ASSERT ic.code is UNIQUE")
graph.run("CREATE CONSTRAINT ON (ua:UserAgent) ASSERT ua.useragent is UNIQUE")
graph.run("CREATE CONSTRAINT ON (m:Method) ASSERT m.method is UNIQUE")
graph.run("CREATE CONSTRAINT ON (r:Referrer) ASSERT r.referrer is UNIQUE")
graph.run("CREATE INDEX ON :DNS(uid)")
graph.run("CREATE INDEX ON :Uri(uri)")
graph.run("CREATE INDEX ON :HTTP(uid)")
也許有人可以給我暗示我做錯了什麼或者我在代碼中犯了什麼錯誤？ 提交的數量是由於嘗試寫入neo4j時的暫時錯誤造成的。隨着更多的交易，我沒有更多的錯誤。
預先感謝任何幫助
您是否製作一些模式約束/索引？ – logisima
我已將限制和索引添加到問題 –