一文速学-知识图谱从零开始构建实战Python指南

原创

fanstuck

修改于 2024-11-28 17:53:55

4391

“好事”文章推荐：大数据挖掘实战-PyODPS基础操作

文章地址：https://cloud.tencent.com/developer/article/2470521

推荐原因：很多第三方平台都有自己的大数据工具以及代码工具库，因此本系列内容就是主要写PyODPS这个目前算是主流常用的大数据类PySpark库，主要依托于阿里云的DataWorks，可以直接在大数据开发MaxCompute使用PyODPS，十分方便数据挖掘。因此本系列将延展MaxComputer来进行一系列数据挖掘项目，有需求的不要错过。

前言

项目目前为止推进到了第五步，也就离我们最终目标自动化实现知识图谱生产不远了。承接上一章的内容，我们本章内容需要联通Python操作Neo4j达到全流程贯通，目前已经有比较成熟的做法，而且我们上篇内容已经完成了包含Neo4j的使用和对数据的基础操作，因此我们很好理解通过Python去使用Neo4j，那么废话不多说现在开始。对知识图谱感兴趣的同学不要错过本人的从零搭建项目专栏文章。

基础操作

安装

需要注意Python环境，需要Python >= 3.7

pip install neo4j

连接

安装驱动程序并运行Neo4j实例后，就可以将应用程序连接到数据库了。通过创建Driver对象并提供URL和身份验证令牌，可以连接到数据库。

from neo4j import GraphDatabase

# URI examples: "neo4j://localhost", "neo4j+s://xxx.databases.neo4j.io"
URI = "<URI for Neo4j database>"
AUTH = ("<Username>", "<Password>")

with GraphDatabase.driver(URI, auth=AUTH) as driver:
    driver.verify_connectivity()
    print("Connection established.")

创建Driver实例仅提供有关如何访问数据库的信息，但实际上并没有建立连接。连接被推迟到执行第一个查询时，要立即验证驱动程序是否可以连接到数据库（有效凭据、兼容版本等），需要在初始化驱动程序后使用.verify_connectivity（）方法。

创建Driver对象和连接验证都会引发许多不同的异常。由于错误处理可能会变得非常冗长，并且连接错误会阻碍任何后续任务，因此最常见的选择是在连接过程中发生异常时让程序崩溃。

驱动程序对象是不可变的、线程安全的，并且创建成本很高，因此应用程序应该只创建一个实例并传递它（可以跨线程共享驱动程序实例）。如果需要通过多个不同的用户查询数据库，可以在另一个用户的安全上下文下执行查询，并使用参数pseudomateduser指定要模拟的用户的名称。为此，创建驱动程序的用户需要具有适当的权限。模拟用户比创建新的Driver对象更优。

driver.execute_query(
    "MATCH (p:Person) RETURN p.name",
    impersonated_user_="somebody_else",
    database_="neo4j",
)

，而无需创建新的Driver实例。如果要更改驱动程序配置，则需要创建一个新对象。

关闭连接

关闭Driver对象以释放所有分配的资源，即使在连接失败或运行时错误的情况下也是如此。使用with语句实例化Driver对象，或者显式调用Driver.close（）方法。

driver.close（）

查询

连接到数据库后，可以使用Cypher和Driver.execute_query（）方法运行查询。Driver.execut_query（）是在5.8版本的驱动程序中引入的。对于早期版本的查询，需要使用sessions and transactions.。

写入数据库

要创建一个代表名为Alice的人的节点，可以使用Cypher子句create：

summary = driver.execute_query(
    "CREATE (:Person {name: $name})",
    name="Alice",
    database_="neo4j",
).summary
print("Created {nodes_created} nodes in {time} ms.".format(
    nodes_created=summary.counters.nodes_created,
    time=summary.result_available_after
))

我们再次输入Cypher就可以看到：

MATCH (tom:Person)
WHERE tom.name = "Alice"
RETURN tom

读取数据

要从数据库中检索信息，和上述在Cypher子句MATCH一样，传递给Neo4j即可：

records, summary, keys = driver.execute_query(
    "MATCH (tom:Person)\
    WHERE tom.name = 'Alice'\
    RETURN tom",
    database_="neo4j",
)

# Loop through results and do something with them
for record in records:
    print(record.data())  # obtain record as dict

# Summary information
print("The query `{query}` returned {records_count} records in {time} ms.".format(
    query=summary.query, records_count=len(records),
    time=summary.result_available_after
))

records以Record对象数组的形式包含结果，summary包含服务器返回的执行摘要。

更新

要更新数据库中的节点信息，可以使用Cypher子句MATCH和SET：

records, summary, keys = driver.execute_query("""
    MATCH (p:Person {name: $name})
    SET p.age = $age
    """, name="Alice", age=42,
    database_="neo4j",
)
print(f"Query counters: {summary.counters}.")

records, summary, keys = driver.execute_query(    "MATCH (people:Person)\    WHERE people.name = 'Alice'\    RETURN people.age",    database_="neo4j",)

要创建一个新的关系，将其链接到两个已经存在的节点，则可以使用Cypher子句MATCH和create的组合：

records, summary, keys = driver.execute_query("""
    MATCH (alice:Person {name: $name})
    MATCH (bob:Person {name: $friend})
    CREATE (alice)-[:KNOWS]->(bob)
    """, name="Alice", friend="Bob",
    database_="neo4j",
)
print(f"Query counters: {summary.counters}.")

如果没有Bob的话可以先创建这个节点：

summary = driver.execute_query(
    "CREATE (:Person {name: $name})",
    name="Bob",
    database_="neo4j",
).summary
print("Created {nodes_created} nodes in {time} ms.".format(
    nodes_created=summary.counters.nodes_created,
    time=summary.result_available_after
))

之后运行上述插入关系显示打印：

插入成功。

删除

要删除节点及其附加的任何关系，需要使用Cypher子句DETACH DELETE：

records, summary, keys = driver.execute_query("""
    MATCH (p:Person {name: $name})
    DETACH DELETE p
    """, name="Bob",
    database_="neo4j",
)
print(f"Query counters: {summary.counters}.")

查询参数

查询写代码时不要将参数直接硬编码或连接到查询中。相反，请始终使用占位符并指定Cypher参数，如前面的示例所示。这是为了：

性能优势：Neo4j编译和缓存查询，但只有在查询结构不变的情况下才能这样做；
安全原因：防止Cypher注入。

查询参数可以作为多个关键字参数传递，也可以在字典中作为parameters_关键字参数的值组合在一起。在混合的情况下，关键字参数优先于字典参数。

parameters = {
    "name": "Alice",
    "age": 42
}
driver.execute_query(
    "MERGE (charlie {name: $name, age: $age})\
    RETURN charlie",
    parameters_=parameters,
    database_="neo4j",
)

错误处理

因为.execut_query（）可能会引发许多不同的异常，处理错误的最佳方法是在单个try/except块中捕获所有异常：

try:
    driver.execute_query(...)
except Exception as e:
    ...  # handle exception

如果故障被认为是暂时的（例如由于临时服务器不可用），驱动程序会自动重试运行失败的查询。如果多次尝试后操作仍然失败，将引发异常。

查询配置

可以提供更多的关键字参数来更改.executquery（）的默认行为，配置参数后缀为。

数据库选择

建议使用database_参数显式指定数据库，即使在单个数据库实例上也是如此。这允许驱动程序更有效地工作，因为它节省了到服务器的网络往返时间来解析本地数据库。如果没有给出数据库，则使用用户的主数据库。

driver.execute_query(
    "MATCH (p:Person) RETURN p.name",
    database_="neo4j",
)

请求路由

在集群环境中，默认情况下，所有查询都会定向到leader节点。为了提高读取查询的性能，可以使用参数routing_=“r”将查询路由到读取节点。

driver.execute_query(
    "MATCH (p:Person) RETURN p.name",
    routing_="r",  # short for neo4j.RoutingControl.READ
    database_="neo4j",
)

虽然在读取模式下执行写查询可能会导致运行时错误，但不应该依赖此来进行访问控制。这两种模式的区别在于，读取事务将被路由到集群的任何节点，而写入事务将被定向到leader。换句话说，不能保证以读取模式提交的写入查询会被拒绝。

其他身份登陆

我们还可以在另一个用户的安全上下文下执行查询，使用参数pseudomateduser指定要模拟的用户的名称。为此创建驱动程序的用户需要具有适当的权限。不需要也不推荐创建新的Driver对象。

driver.execute_query(
    "MATCH (p:Person) RETURN p.name",
    impersonated_user_="somebody_else",
    database_="neo4j",
)

模拟用户时，查询在模拟用户的完整安全上下文中运行，而不是在经过身份验证的用户（即主数据库、权限等）中运行。

转换查询结果

可以使用resulttransformer参数将查询的结果转换为不同的数据结构。驱动程序提供了将结果转换为pandas Dataframe或图形的内置方法，也可以制作自己的转换器。

完整创建案例

那么我们综合上述来使用一个完整的创建图案例：

from neo4j import GraphDatabase


URI = "bolt://localhost:7687"
AUTH = ("neo4j", "xwt353008")

people = [{"name": "Alice", "age": 42, "friends": ["Bob", "Peter", "Anna"]},
          {"name": "Bob", "age": 19},
          {"name": "Peter", "age": 50},
          {"name": "Anna", "age": 30}]

with GraphDatabase.driver(URI, auth=AUTH) as driver:
    try:
        # Create some nodes
        for person in people:
            records, summary, keys = driver.execute_query(
                "MERGE (p:Person {name: $person.name, age: $person.age})",
                person=person,
                database_="neo4j",
            )

        # Create some relationships
        for person in people:
            if person.get("friends"):
                records, summary, keys = driver.execute_query("""
                    MATCH (p:Person {name: $person.name})
                    UNWIND $person.friends AS friend_name
                    MATCH (friend:Person {name: friend_name})
                    MERGE (p)-[:KNOWS]->(friend)
                    """, person=person,
                    database_="neo4j",
                )

        # Retrieve Alice's friends who are under 40
        records, summary, keys = driver.execute_query("""
            MATCH (p:Person {name: $name})-[:KNOWS]-(friend:Person)
            WHERE friend.age < $age
            RETURN friend
            """, name="Alice", age=40,
            routing_="r",
            database_="neo4j",
        )
        # Loop through results and do something with them
        for record in records:
            print(record)
        # Summary information
        print("The query `{query}` returned {records_count} records in {time} ms.".format(
            query=summary.query, records_count=len(records),
            time=summary.result_available_after
        ))

    except Exception as e:
        print(e)
        # further logging/processing