图数据库｜如何从零到一构建一个企业股权图谱系统

GO FROM "p_100" OVER hold_share YIELD dst(edge) AS corp_with_share |\GO FROM $-.corp_with_share OVER hold_share REVERSELY YIELD properties(vertex).name;

SELECT a.id, a.name, c.nameFROM person aJOIN hold_share b ON a.id=b.person_idJOIN corp c ON c.id=b.corp_idWHERE c.name IN (SELECT c.nameFROM person aJOIN hold_share b ON a.id=b.person_idJOIN corp c ON c.id=b.corp_idWHERE a.id = 'p_100')

$ pip install Faker==2.0.5 pydbgen==1.0.5$ python3 data_generator.py$ ls -l datatotal 1688-rw-r--r--  1 weyl  staff   23941 Jul 14 13:28 corp.csv-rw-r--r--  1 weyl  staff    1277 Jul 14 13:26 corp_rel.csv-rw-r--r--  1 weyl  staff    3048 Jul 14 13:26 corp_share.csv-rw-r--r--  1 weyl  staff  211661 Jul 14 13:26 person.csv-rw-r--r--  1 weyl  staff  179770 Jul 14 13:26 person_corp_role.csv-rw-r--r--  1 weyl  staff  322965 Jul 14 13:26 person_corp_share.csv-rw-r--r--  1 weyl  staff   17689 Jul 14 13:26 person_rel.csv

git clone https://github.com/wey-gu/nebula-shareholding-examplecp -r data_sample /tmp/datacp nebula-importer.yaml /tmp/data/docker run --rm -ti \    --network=nebula-docker-compose_nebula-net \    -v /tmp/data:/root \    vesoft/nebula-importer:v2.6.0 \    --config /root/nebula-importer.yaml

CREATE SPACE IF NOT EXISTS shareholding(partition_num=5, replica_factor=1, vid_type=FIXED_STRING(10));USE shareholding;CREATE TAG person(name string);CREATE TAG corp(name string);CREATE TAG INDEX person_name on person(name(20));CREATE TAG INDEX corp_name on corp(name(20));CREATE EDGE role_as(role string);CREATE EDGE is_branch_of();CREATE EDGE hold_share(share float);CREATE EDGE reletive_with(degree int);

  ┌───────────────┬───────────────┐  │               │  Frontend     │  │               │               │  │    ┌──────────▼──────────┐    │  │    │ Vue.JS              │    │  │    │ D3.JS               │    │  │    └──────────┬──────────┘    │  │               │  Backend      │  │    ┌──────────┴──────────┐    │  │    │ Flask               │    │  │    │ Nebula-Python       │    │  │    └──────────┬──────────┘    │  │               │  Graph Query  │  │    ┌──────────▼──────────┐    │  │    │ Graph Database      │    │  │    └─────────────────────┘    │  │                               │  └───────────────────────────────┘

MATCH p=(v)-[e:hold_share|:is_branch_of|:reletive_with|:role_as*1..3]-(v2) \WHERE id(v) IN ["c_132"] RETURN p LIMIT 100

$ python3 -m pip install nebula2-python==2.5.0 # 注意这里我引用旧的记录，它是 2.5.0，$ ipythonIn [1]: from nebula2.gclient.net import ConnectionPoolIn [2]: from nebula2.Config import ConfigIn [3]: config = Config()   ...: config.max_connection_pool_size = 10   ...: # init connection pool   ...: connection_pool = ConnectionPool()   ...: # if the given servers are ok, return true, else return false   ...: ok = connection_pool.init([('192.168.8.137', 9669)], config)   ...: session = connection_pool.get_session('root', 'nebula')[2021-10-13 13:44:24,242]:Get connection to ('192.168.8.137', 9669)
In [4]: resp = session.execute("use shareholding")In [5]: query = '''   ...: MATCH p=(v)-[e:hold_share|:is_branch_of|:reletive_with|:role_as*1..3]-(v2) \   ...: WHERE id(v) IN ["c_132"] RETURN p LIMIT 100   ...: '''In [6]: resp = session.execute(query) # Note: after nebula graph 2.6.0, we could use execute_json as well
In [7]: resp.col_size()Out[7]: 1
In [9]: resp.row_size()Out[10]: 100

In [11]: p=resp.row_values(22)[0].as_path()
In [12]: p.nodes()Out[12]:[("c_132" :corp{name: "Chambers LLC"}), ("p_4000" :person{name: "Colton Bailey"})]
In [13]: p.relationships()Out[13]: [("p_4000")-[:role_as@0{role: "Editorial assistant"}]->("c_132")]

In [14]: rel=p.relationships()[0]
In [15]: relOut[15]: ("p_4000")-[:role_as@0{role: "Editorial assistant"}]->("c_132")
In [16]: rel.edge_name()Out[16]: 'role_as'
In [17]: rel.properties()Out[17]: {'role': "Editorial assistant"}
In [18]: rel.start_vertex_id()Out[18]: "p_4000"
In [19]: rel.end_vertex_id()Out[19]: "c_132"

In [20]: node=p.nodes()[0]
In [21]: node.tags()Out[21]: ['corp']
In [22]: node.properties('corp')Out[22]: {'name': "Chambers LLC"}
In [23]: node.get_id()Out[23]: "c_132"

nodes: [        {"id": "c_132", "name": "Chambers LLC", "tag": "corp"},        {"id": "p_4000", "name": "Colton Bailey", "tag": "person"}],relationships: [        {"source": "p_4000", "target": "c_132", "properties": { "role": "Editorial assistant" }, "edge": "role_as"}]

[{"id": "c_132", "name": "Chambers LLC", "tag": "corp"}, {"id": "p_4000", "name": "Colton Bailey", "tag": "person"}]

[{"source": "p_4000", "target": "c_132", "properties": { "role": "Editorial assistant" }, "edge": "role_as"}, {"source": "p_1039", "target": "c_132", "properties": { "share": "3.0" }, "edge": "hold_share"}]

def make_graph_response(resp) -> dict:    nodes, relationships = list(), list()    for row_index in range(resp.row_size()):        path = resp.row_values(row_index)[0].as_path()        _nodes = [            {                "id": node.get_id(), "tag": node.tags()[0],                "name": node.properties(node.tags()[0]).get("name", "")                }                for node in path.nodes()        ]        nodes.extend(_nodes)        _relationships = [            {                "source": rel.start_vertex_id(),                "target": rel.end_vertex_id(),                "properties": rel.properties(),                "edge": rel.edge_name()                }                for rel in path.relationships()        ]        relationships.extend(_relationships)    return {"nodes": nodes, "relationships": relationships}

from flask import Flask, jsonify, request


app = Flask(__name__)

@app.route("/")def root():    return "Hey There?"

@app.route("/api", methods=["POST"])def api():    request_data = request.get_json()    entity = request_data.get("entity", "")    if entity:        resp = query_shareholding(entity)        data = make_graph_response(resp)    else:        data = dict() # tbd    return jsonify(data)

def parse_nebula_graphd_endpoint():    ng_endpoints_str = os.environ.get(        'NG_ENDPOINTS', '127.0.0.1:9669,').split(",")    ng_endpoints = []    for endpoint in ng_endpoints_str:        if endpoint:            parts = endpoint.split(":")  # we dont consider IPv6 now            ng_endpoints.append((parts[0], int(parts[1])))    return ng_endpoints
def query_shareholding(entity):    query_string = (        f"USE shareholding; "        f"MATCH p=(v)-[e:hold_share|:is_branch_of|:reletive_with|:role_as*1..3]-(v2) "        f"WHERE id(v) IN ['{ entity }'] RETURN p LIMIT 100"    )    session = connection_pool.get_session('root', 'nebula')    resp = session.execute(query_string)    return resp

curl --header "Content-Type: application/json" \     --request POST \     --data '{"entity": "c_132"}' \     http://192.168.10.14:5000/api | jq
{  "nodes": [    {      "id": "c_132",      "name": "\"Chambers LLC\"",      "tag": "corp"    },    {      "id": "c_245",      "name": "\"Thompson-King\"",      "tag": "corp"    },    {      "id": "c_132",      "name": "\"Chambers LLC\"",      "tag": "corp"    },...    }  ],  "relationships": [    {      "edge": "hold_share",      "properties": "{'share': 0.0}",      "source": "c_245",      "target": "c_132"    {      "edge": "hold_share",      "properties": "{'share': 9.0}",      "source": "p_1767",      "target": "c_132"    },    {      "edge": "hold_share",      "properties": "{'share': 11.0}",      "source": "p_1997",      "target": "c_132"    },...    },    {      "edge": "reletive_with",      "properties": "{'degree': 51}",      "source": "p_7283",      "target": "p_4723"    }  ]}

├── README.md         # You could find Design Logs here├── corp-rel-backend│   └── app.py        # Flask App to handle Requst and calls GDB├── corp-rel-frontend│   └── src│       ├── App.vue│       └── main.js   # Vue App to call Flask App and Renders Graph└── requirements.txt