图数据库 Neo4j 的部署、数据导入和简单使用

使用 neo4j-import 命令行 工具 导入 csv 数据是几种数据加载方式中最快的一种，但它不能导入数据到已有的数据库中，每次执行都是产生一个全新的数据库，因此必须在一条命令里将数据库中要包含的数据全部都制定好。

可以用下面的命令来导入豆瓣电影图谱数据

neo4j-import --into graph.db --id-type string \
             --nodes:Person movie/Person.csv    \
             --nodes:Movie movie/Movie.csv \
             --nodes:Country movie/Country.csv \
             --relationships:actor movie/actor.csv \
             --relationships:composer movie/composer.csv \
             --relationships:director movie/director.csv \
             --relationships:district movie/district.csv

上述命令会在当前目录下生成一个 graph.db 目录，就是最终产生的一个全新的数据库。要启用这个数据库，必须将其放置到 Neo4j Server 的 data 目录下：

• 如果当前 Neo4j Server 正在运行，需要先停掉它

  neo4j stop

• 删除或备份原有的数据库

  mv /opt/neo4j/data/databases/graph.db /opt/neo4j/data/databases/graph.db.bak

• 将产生的 graph.db 放置到 server 的 data 目录下

  cp graph.db /opt/neo4j/data/databases/ -r

• 重新启动 Neo4j Server

  neo4j start

实体和关系一共 8 万多条，在我的个人电脑上一共花费 3s 多

IMPORT DONE in 3s 692ms.
Imported:
  27608 nodes
  54844 relationships
  91628 properties
Peak memory usage: 524.24 MB

如果是以 docker 的方式来使用 Neo4j，则稍有不同，需要在执行的时候将 movie 目录和输出结果所在的目录都挂载到容器里。假设说我们希望最终输出结果到 $HOME/neo4j/data 目录下，那么，先创建这个目录

mkdir $HOME/neo4j/data/databases -p

然后执行

docker run -v $PWD/movie:/movie:ro -v $HOME/neo4j/data:/data/ \
       neo4j neo4j-import --into /data/databases/graph.db --id-type string \
             --nodes:Person /movie/Person.csv    \
             --nodes:Movie /movie/Movie.csv \
             --nodes:Country /movie/Country.csv \
             --relationships:actor /movie/actor.csv \
             --relationships:composer /movie/composer.csv \
             --relationships:director /movie/director.csv \
             --relationships:district /movie/district.csv

然后再用 docker 启动 Neo4j Server，并让其使用刚刚产生的数据库

docker run -p 7474:7474 \
       -p 7687:7687 \
       -v $HOME/neo4j/data/:/data \
       -e NEO4J_AUTH=neo4j/neo4j_password \
       neo4j

用 LOAD CSV 语句同样可以加载 csv 数据，不过和 neo4j-import 不一样，本质上它只是负责从 csv 文件中读取数据，如果要将读取到的数据写入到数据库中，还必须通过 CREATE 语句。也正因如此，用 LOAD CSV 语句来加载数据，不需要将 Neo4j Server 停掉。

用 LOAD CSV 语句将豆瓣电影图谱加载到数据库中的做法是下面这样的

• 从 Movie.csv 中加载电影数据并创建 Movie 实体

  USING PERIODIC COMMIT 1000
  LOAD CSV with headers from 'https://raw.githubusercontent.com/Linusp/kg-example/master/movie/Movie.csv' as line
  CREATE (:Movie {
         id:line["id:ID"],
         title:line["title"],
         url:line["url"],
         cover:line["cover"],
         rate:line["rate"],
         category:split(line["category:String[]"], ";"),
         language:split(line["language:String[]"], ";"),
         showtime:line["showtime"],
         length:line["length"],
         othername:split(line["othername:String[]"], ";")
         })

  其中 "using periodic commit 1000" 表示每读取 1000 行数据就写入一次。

• 从 Person.csv 中加载人员数据并创建 Person 实体

  USING PERIODIC COMMIT 1000
  LOAD CSV with headers from 'https://raw.githubusercontent.com/Linusp/kg-example/master/movie/Person.csv' as line
  CREATE (:Person {id:line["id:ID"], name:line["name"]})

• 从 Country.csv 中加载国家数据并创建 Country 实体

  USING PERIODIC COMMIT 1000
  LOAD CSV with headers from 'https://raw.githubusercontent.com/Linusp/kg-example/master/movie/Country.csv' as line
  CREATE (:Country {id:line["id:ID"], name:line["name"]})

• 创建关系

  每个关系的 csv 文件都是如下格式（以 actor.csv 为例）

  ":START_ID",":END_ID"
  "5ec851a8b7b7bbf0c9f42bbee021be00","3a20ded16ebce312f56a562e1bef7f05"
  "5ec851a8b7b7bbf0c9f42bbee021be00","8101549e05e6c1afbea62890117c01c6"
  "5ec851a8b7b7bbf0c9f42bbee021be00","111a3c7f6b769688da55828f36bbd604"
  "5ec851a8b7b7bbf0c9f42bbee021be00","5cc5d969f42ce5d8e3937e37d77b89b5"
  "5ec851a8b7b7bbf0c9f42bbee021be00","a5e6012efc56f0ca07184b9b88eb2373"
  "5ec851a8b7b7bbf0c9f42bbee021be00","435c8172c14c24d6cd123c529a0c2a76"
  "5ec851a8b7b7bbf0c9f42bbee021be00","5dfb355a385bcfe9b6056b8d322bfecb"
  "5ec851a8b7b7bbf0c9f42bbee021be00","5076a2f7479462dcc4637b6fe3226095"
  "5ec851a8b7b7bbf0c9f42bbee021be00","c7103a9ad17cf56fd572657238e49fff"

在创建关系的时候实际上是根据两个 id 查询到对应的实体，然后再为其建立关系。虽然我在准备这份数据时，已经保证了每个实体的 id 都是全局唯一的，但在没有创建索引的情况下，用这个 id 来查询实体会以遍历的形式进行，效率很差，所以在创建关系前，先创建一下索引。

为 Movie 实体的 id 属性创建索引

CREATE INDEX ON :Movie(id)

为 Person 实体的 id 属性创建索引

CREATE INDEX ON :Person(id)

为 Country 实体的 id 属性创建索引

CREATE INDEX ON :Country(id)

然后继续用 LOAD CSV 来创建关系

• 从 actor.csv 中加载数据并创建 actor 关系

  USING PERIODIC COMMIT 1000
  LOAD CSV with headers from 'https://raw.githubusercontent.com/Linusp/kg-example/master/movie/actor.csv' as line
  MATCH (a:Movie {id:line[":START_ID"]})
  MATCH (b:Person {id:line[":END_ID"]})
  CREATE (a)-[:actor]->(b)

• 从 composer.csv 中加载数据创建 composer 关系

  USING PERIODIC COMMIT 1000
  LOAD CSV with headers from 'https://raw.githubusercontent.com/Linusp/kg-example/master/movie/composer.csv' as line
  MATCH (a:Movie {id:line[":START_ID"]})
  MATCH (b:Person {id:line[":END_ID"]})
  CREATE (a)-[:composer]->(b)

• 从 director.csv 中加载数据创建 director 关系

  USING PERIODIC COMMIT 1000
  LOAD CSV with headers from 'https://raw.githubusercontent.com/Linusp/kg-example/master/movie/director.csv' as line
  MATCH (a:Movie {id:line[":START_ID"]})
  MATCH (b:Person {id:line[":END_ID"]})
  CREATE (a)-[:director]->(b)

• 从 district.csv 中加载数据并创建 district 关系

  USING PERIODIC COMMIT 1000
  LOAD CSV with headers from 'https://raw.githubusercontent.com/Linusp/kg-example/master/movie/district.csv' as line
  MATCH (a:Movie {id:line[":START_ID"]})
  MATCH (b:Country {id:line[":END_ID"]})
  CREATE (a)-[:district]->(b)

严格来说，上面的 LOAD CSV 的方式，也是在用 Cypher 语句，不过说到底它还是要依赖一个外部的 CSV 文件，自由度没那么高。而 Neo4j Server 本身还提供 RESTful API，利用这个 API 就可以进行编程来完成更复杂的需求。

以创建实体为例来说明一下 Neo4j Server 的 RESTful API。假设说我们要创建三个 Person 实体，简单起见，我们假设每个 Person 实体需要有 id, name, age 三个属性，比如

[
    {
        "id": "person1",
        "name": "张志昂",
        "age": 23
    },
    {
        "id": "person2",
        "name": "刘文刀",
        "age": 18
    },
    {
        "id": "person3",
        "name": "孙子小",
        "age": 22
    }
]

通过 RESTful API，可以一次性创建这三个 Person 实体

POST http://neo4j:neo4j_password@localhost:7474/db/data/cypher
Content-Type: application/json
{
    "query": "UNWIND {values} as data CREATE (:Person {id: data.id, name: data.name, age: data.age})",
    "params": {
        "values": [
            {"id": "person1", "name": "张志昂", "age": 23},
            {"id": "person2", "name": "刘文刀", "age": 18},
            {"id": "person3", "name": "孙子小", "age": 22}
        ]
    }
}

这种通过带参数的 query 进行批量写入的方式，和 MySQL 等数据库的接口很相似，不过在 Cypher 中可以通过 UNWIND 语句做一些复杂的事情。详见 文档 。

用 Python 来做的话大概是这个样子

import requests

url = "http://neo4j:neo4j_password@localhost:7474/db/data/cypher"
payload = {
    "query": (
        "UNWIND {values} as data "
        "CREATE (:Person {id: data.id, name: data.name, age: data.age})"
    ),
    "params": {
        "values": [
            {"id": "person1", "name": "张志昂", "age": 23},
            {"id": "person2", "name": "刘文刀", "age": 18},
            {"id": "person3", "name": "孙子小", "age": 22}
        ]
    }
}
requests.post(url, json=payload)

或者也可以使用 Neo4j 官方的 Python 客户端

import neo4j

client = neo4j.GraphDatabase.driver(
    'bolt://localhost:7687', auth=('neo4j', 'neo4j_password')
)
with client.session() as session:
    query = (
        "UNWIND {values} as data "
        "create (:Person {id: data.id, name: data.name, age: data.age})"
    )
    values = [
        {"id": "person1", "name": "张志昂", "age": 23},
        {"id": "person2", "name": "刘文刀", "age": 18},
        {"id": "person3", "name": "孙子小", "age": 22}
    ]
    session.run(query, {'values': values})