Skip to content

Commit 3244980

Browse files
gitbudagitbuda
authored
Port to Memgraph 2.0 features (#27)
1 parent 22fb328 commit 3244980

File tree

17 files changed

+224
-167
lines changed

17 files changed

+224
-167
lines changed

‎README.md

+6-5
Original file line numberDiff line numberDiff line change
@@ -14,25 +14,26 @@
1414

1515
This repository serves as a point of reference when developing a streaming application with [Memgraph](https://memgraph.com) and a message broker such as [Kafka](https://kafka.apache.org).
1616

17-
![drawing](https://i.imgur.com/5YMlN8M.png)
17+
![Example Streaming App](https://user-images.githubusercontent.com/4950251/137717495-fab38a69-b087-44ef-90b4-188a7187fbab.png)
1818

1919
*KafkaProducer* represents the source of your data.
2020
That can be transactions, queries, metadata or something different entirely.
2121
In this minimal example we propose using a [special string format](./kafka) that is easy to parse.
2222
The data is sent from the *KafkaProducer* to *Kafka* under a topic aptly named *topic*.
2323
The *Backend* implements a *KafkaConsumer*.
24-
It takes data from *Kafka*, parses it and sends it to *Memgraph* for graph analysis, feature extraction or storage.
24+
It takes data from *Kafka*, consumes it, but also queries *Memgraph* for graph analysis, feature extraction or storage.
2525

2626
## Installation
2727
Install [Kafka](./kafka) and [Memgraph](./memgraph) using the instructions in the homonymous directories.
2828
Then choose a programming language from the list of supported languages and follow the instructions given there.
2929

3030
### List of supported programming languages
31+
- [c#](./backend/cs)
32+
- [go](./backend/go)
33+
- [java](./backend/java)
3134
- [node](./backend/node)
3235
- [python](./backend/python)
33-
- [java](./backend/java)
34-
- [go](./backend/go)
35-
- [c#](./backend/cs)
36+
- [rust](./backend/rust)
3637

3738
## How does it work *exactly*
3839
### KafkaProducer

‎backend/cs/memgraph-streaming/Program.cs

+5-27
Original file line numberDiff line numberDiff line change
@@ -8,12 +8,6 @@ class Program
88
{
99
static void Main(string[] args)
1010
{
11-
var cypherNodeCommand = "MERGE (node:{0} {1}) "
12-
+ "SET node += {2}";
13-
var cypherEdgeCommand = "MERGE (node1:{0} {1}) "
14-
+ "MERGE (node2:{2} {3}) "
15-
+ "MERGE (node1)-[:{4} {5}]->(node2)";
16-
1711
using var driver = GraphDatabase.Driver("bolt://localhost:7687", AuthTokens.None);
1812
using var session = driver.Session();
1913

@@ -30,32 +24,16 @@ static void Main(string[] args)
3024
var message = consumer.Consume().Message.Value;
3125
System.Console.WriteLine("received message: " + message);
3226
var arr = message.Split("|");
33-
var cypherCommand = "";
34-
switch (arr[0])
35-
{
36-
case "node":
37-
cypherCommand = string.Format(cypherNodeCommand, arr[1], arr[2], arr[3]);
38-
break;
39-
case "edge":
40-
cypherCommand = string.Format(cypherEdgeCommand, arr[1], arr[2], arr[5], arr[6], arr[3], arr[4]);
41-
break;
42-
default:
43-
throw new InvalidOperationException(
44-
string.Format("Command '{0}' not supported.", message)
45-
);
46-
}
47-
System.Console.WriteLine(cypherCommand);
48-
session.WriteTransaction(tx =>
49-
{
50-
tx.Run(cypherCommand);
51-
return "";
52-
});
5327
if (arr[0] == "node") {
5428
var neighbors = session.WriteTransaction(tx =>
5529
{
5630
return tx.Run(string.Format("MATCH (node:{0} {1}) RETURN node.neighbors AS neighbors", arr[1], arr[2])).Peek();
5731
});
58-
Console.WriteLine(string.Format("Node (node:{0} {1}) has {2} neighbors.", arr[1], arr[2], neighbors.Values["neighbors"]));
32+
if (neighbors != null) {
33+
Console.WriteLine(string.Format("Node (node:{0} {1}) has {2} neighbors.", arr[1], arr[2], neighbors.Values["neighbors"]));
34+
} else {
35+
Console.WriteLine("Neighbors number is null. Triggers are not defined or not yet executed.");
36+
}
5937
}
6038
}
6139
}

‎backend/cs/memgraph-streaming/memgraph-streaming.csproj

+2-2
Original file line numberDiff line numberDiff line change
@@ -7,8 +7,8 @@
77
</PropertyGroup>
88

99
<ItemGroup>
10-
<PackageReference Include="Confluent.Kafka" Version="1.7.0" />
11-
<PackageReference Include="Neo4j.Driver.Simple" Version="4.3.1" />
10+
<PackageReference Include="Confluent.Kafka" Version="1.8.1" />
11+
<PackageReference Include="Neo4j.Driver.Simple" Version="4.3.2" />
1212
</ItemGroup>
1313

1414
</Project>

‎backend/go/app.go

-21
Original file line numberDiff line numberDiff line change
@@ -17,43 +17,22 @@ func main() {
1717
}
1818
defer driver.Close()
1919

20-
cypherNodeCommand := "MERGE (node:%s %s) " +
21-
"SET node += %s"
22-
cypherEdgeCommand := "MERGE (node1:%s %s) " +
23-
"MERGE (node2:%s %s) " +
24-
"MERGE (node1)-[:%s %s]->(node2)"
25-
2620
kafkaReader := kafka.NewReader(kafka.ReaderConfig{
2721
Brokers: []string{"localhost:9092"},
2822
Topic: "topic",
2923
MinBytes: 0,
3024
MaxBytes: 10e6,
3125
})
3226
defer kafkaReader.Close()
33-
kafkaLoop:
3427
for {
3528
kafkaMessage, err := kafkaReader.ReadMessage(context.Background())
3629
if err != nil {
3730
fmt.Println("nothing to read...")
3831
break
3932
}
4033
message := string(kafkaMessage.Value)
41-
cypherCommand := ""
4234
arr := strings.Split(message, "|")
4335

44-
switch arr[0] {
45-
case "node":
46-
cypherCommand = fmt.Sprintf(cypherNodeCommand, arr[1], arr[2], arr[3])
47-
case "edge":
48-
cypherCommand = fmt.Sprintf(cypherEdgeCommand, arr[1], arr[2], arr[5], arr[6], arr[3], arr[4])
49-
default:
50-
fmt.Printf("invalid kafka message: `%s`", message)
51-
break kafkaLoop
52-
}
53-
_, err = runCypherCommand(driver, cypherCommand)
54-
if err != nil {
55-
panic(err)
56-
}
5736
if arr[0] == "node" {
5837
result, err := runCypherCommand(
5938
driver,

‎backend/java/memgraph-streaming/src/main/java/memgraph/App.java

-16
Original file line numberDiff line numberDiff line change
@@ -16,11 +16,6 @@
1616

1717
public class App {
1818
public static void main(String[] args) throws Exception {
19-
String nodeQuery = "MERGE (node:%s %s) "
20-
+ "SET node += %s";
21-
String edgeQuery = "MERGE (node1:%s %s) "
22-
+ "MERGE (node2:%s %s) "
23-
+ "MERGE (node1)-[:%s %s]->(node2)";
2419

2520
try (Driver driver = GraphDatabase.driver("bolt://localhost:7687");
2621
Session session = driver.session();
@@ -38,24 +33,13 @@ public static void main(String[] args) throws Exception {
3833
public String execute(Transaction tx) {
3934
switch (command[0]) {
4035
case "node":
41-
tx.run(String.format(nodeQuery, command[1], command[2],
42-
command[3]));
4336
Result result = tx.run(String.format(
4437
"MATCH (node:%s %s) RETURN node.neighbors AS neighbors",
4538
command[1], command[2]));
4639
System.out.printf("Node (node:%s %s) has %d neighbors.\n",
4740
command[1], command[2],
4841
result.single().get(0).asInt());
4942
break;
50-
case "edge":
51-
tx.run(String.format(edgeQuery, command[1], command[2],
52-
command[5], command[6], command[3],
53-
command[4]));
54-
break;
55-
default:
56-
System.out.printf("Error: unknown command `%s`\n",
57-
record.value());
58-
return null;
5943
}
6044
System.out.printf("%s\n", record.value());
6145
return null;

‎backend/node/src/index.js

+1-25
Original file line numberDiff line numberDiff line change
@@ -1,17 +1,5 @@
11
/**
2-
* This is a generic Kafka Consumer that will enable you to store data from
3-
* Kafka to Memgraph.
4-
*
5-
* The data pushed to Kafka has to be in the following format:
6-
* command|label|unique_fields|fields
7-
* command|label1|unique_fields1|edge_type|edge_fields|label2|unique_fields2
8-
*
9-
* command - string: "edge", or "node"
10-
* label - string: type(s) of a node e.g. "Person", or "Machine:Vehicle:Car"
11-
* edge_type - string: type of an edge e.g. "CONNECTED_WITH"
12-
* fields - string in form of a json/python dictionary representing the
13-
* properties of a node or edge:
14-
* `{age: 53}` or `{id: 4, name: "hero", alive: true}`
2+
* This is a generic Kafka Consumer + an example express app.
153
*/
164
const express = require('express');
175
const app = express();
@@ -56,18 +44,6 @@ async function runConsumer() {
5644
\n - partition ${partition} \
5745
\n - offset ${offset}. \
5846
\nUpdated total count to ${++kafkaCounter}`);
59-
const [type, ...rest] = value.toString().split('|');
60-
const session = driver.session();
61-
if (type === 'node') {
62-
const [label, uniqueFields, fields] = rest;
63-
await session.run(`MERGE (n:${label} ${uniqueFields}) SET n += ${fields};`);
64-
} else if (type === 'edge') {
65-
const [n1l, n1u, edgeType, edgeFields, n2l, n2u] = rest;
66-
await session.run(`MERGE (n1:${n1l} ${n1u}) MERGE (n2:${n2l} ${n2u}) \
67-
MERGE (n1)-[:${edgeType} ${edgeFields}]->(n2);`);
68-
} else {
69-
throw new Error('Unknown message type.');
70-
}
7147
}),
7248
);
7349
}

‎backend/python/app.py

+13-48
Original file line numberDiff line numberDiff line change
@@ -1,48 +1,26 @@
1+
"""This is a generic Kafka Consumer and an example Python code on how to query
2+
Memgraph.
13
"""
2-
This is a generic Kafka Consumer that will enable you to store data from Kafka
3-
to Memgraph.
4-
The data pushed to Kafka has to be in the following format:
5-
command|label|unique_fields|fields
6-
command|label1|unique_fields1|edge_type|edge_fields|label2|unique_fields2
7-
8-
command - string: "edge", or "node"
9-
label - string: type(s) of a node e.g. "Person", or "Machine:Vehicle:Car"
10-
edge_type - string: type of an edge e.g. "CONNECTED_WITH"
11-
fields - string in form of a json/python dictionary representing the
12-
properties of a node or edge:
13-
`{age: 53}` or `{id: 4, name: "hero", alive: true}`
14-
"""
15-
import csv
164
import logging
5+
import csv
176

187
from gqlalchemy import Memgraph
198
from kafka import KafkaConsumer
209

2110

2211
def process(message: str, db: Memgraph):
23-
"""Takes graph database `db` and a string message in the following format:
24-
25-
command|label|unique_fields|fields
26-
command|label1|unique_fields1|edge_type|edge_fields|label2|unique_fields2
27-
28-
command - string: "edge", or "node"
29-
label - string: type of a node e.g. "Person" or "Machine:Vehicle:Car"
30-
edge_type - string: type of an edge e.g. "CONNECTED_WITH"
31-
fields - string in form of a json/python dictionary representing the
32-
properties of a node or edge:
33-
`{age: 53}` or `{id: 4, name: "hero", alive: true}`
34-
35-
Throws a ValueError if the command isn't recognised.
36-
"""
12+
"""Prints the number of neighbors."""
13+
logging.info(f"Received `{message}`")
3714
payload = next(csv.reader([message], delimiter="|"))
3815
command, *payload = payload
3916

4017
if command == "node":
4118
label, unique_fields, fields = payload
42-
db.execute_query(f"merge (a:{label} {unique_fields}) set a += {fields}")
43-
neighbors = next(db.execute_and_fetch(
44-
f"match (a:{label} {unique_fields}) return a.neighbors as n"
45-
))['n']
19+
neighbors = next(
20+
db.execute_and_fetch(
21+
f"match (a:{label} {unique_fields}) return a.neighbors as n"
22+
)
23+
)["n"]
4624
if neighbors is None:
4725
print(
4826
"The neighbors variable isn't set. "
@@ -51,32 +29,20 @@ def process(message: str, db: Memgraph):
5129
else:
5230
print(f"(node:{label} {unique_fields}) has {neighbors} neighbors.")
5331
elif command == "edge":
54-
(
55-
label1,
56-
unique_fields1,
57-
edge_type,
58-
edge_fields,
59-
label2,
60-
unique_fields2,
61-
) = payload
62-
db.execute_query(
63-
f"merge (a:{label1} {unique_fields1}) "
64-
f"merge (b:{label2} {unique_fields2}) "
65-
f"merge (a)-[:{edge_type} {edge_fields}]->(b)"
66-
)
32+
pass
6733
else:
6834
raise ValueError(f"Command `{command}` not recognized.")
69-
logging.info(f"`{message}`, Successfully entered {command} in Memgraph.")
7035

7136

7237
if __name__ == "__main__":
7338
logging.basicConfig(
74-
filename="info.log",
7539
level=logging.INFO,
7640
format="%(levelname)s: %(asctime)s %(message)s",
7741
)
42+
7843
db = Memgraph(host="localhost", port=7687)
7944
db.drop_database()
45+
8046
consumer = KafkaConsumer("topic", bootstrap_servers=["localhost:9092"])
8147
try:
8248
for message in consumer:
@@ -86,6 +52,5 @@ def process(message: str, db: Memgraph):
8652
except Exception as error:
8753
logging.error(f"`{message}`, {repr(error)}")
8854
continue
89-
9055
except KeyboardInterrupt:
9156
pass

0 commit comments

Comments
 (0)