利用图数据库研究 COVID-19 论文数据集

COVID-19 大流行的形势依然很严峻，为应对 COVID-19 的传播及其对我们的影响，AI2等提供了一份 COVID-19 开放研究数据集(CORD-19)。CORD-19 数据集是关于冠状病毒的文献集，提供了超过50万篇学术论文的相关信息。我们研究了这份数据集，并用图数据库去组织和挖掘这份数据集蕴含的信息。

针对 CORD-19,我们设计了以下的图模型。

图模型中包含了两类点，分别是 paper 和 author，也包含了两类边，分别是 write 和 reference，下面的表格说明了它们的详细情况。

数据集

原始数据集位于COVID-19 开放研究数据集，根据我们设计的图模型重新组织的数据可从此处下载，这份图数据包含70万+ paper 点，173万+ author 点，67万+ reference 边，443万+ write 边。

创图

利用华为云图数据库 GES 对以上数据集进行探索，需要先在 GES 上创图，GES 创图的详细流程见华为云图引擎服务 GES 实战——创图。

使用 GES 查询的预置条件

下面封装了使用 GES 查询的预置条件，包括配置相关参数和对所调用 API 接口的封装，如果你对这些不感兴趣，可直接运行而不需要了解细节，这对理解后续具体查询没有影响。

配置相关参数的定义。

import requests
import json
import time
import pandas as pd
pd.set_option('display.max_rows', None)
pd.set_option('display.max_columns', None)
pd.set_option('display.width', None)
pd.set_option('display.max_colwidth', -1)

class config(object):
    def __init__(self, iam_url, user_name, password, domain_name, project_name, eip, project_id, graph_name):
        self.iam_url = iam_url
        self.user_name = user_name
        self.password = password
        self.domain_name = domain_name
        self.project_name = project_name
        self.eip = eip
        self.project_id = project_id
        self.graph_name = graph_name
        self.token = self.get_token()

    def get_token(self):
        url = ('https://{}/v3/auth/tokens').format(self.iam_url)
        headers = {'Content-Type': 'application/json;charset=utf8'}
        body = json.dumps({"auth": {
            "identity": {
                "methods": ["password"],
                "password": {
                    "user": {
                        "name": self.user_name,
                        "password": self.password,
                        "domain": {
                            "name": self.domain_name
                        }
                    }
                }
            },
            "scope": {
                "project": {
                    "name": self.project_name
                }
            }
        }
        })
        r = requests.post(url=url, data=body, verify=False, headers=headers)
        return r.headers['x-subject-token']

config类的参数都是与调用 GES 服务有关的参数，依次为“终端节点”、“IAM 用户名”、“IAM 用户密码”、“IAM 用户所属账户名”、“项目名称”、“公网访问地址”、“项目ID”、“token值”，其获取方式可参考调用 GES 服务业务面 API 相关参数的获取。

下面定义了与查询相关的接口。

class GESFunc(object):

    def __init__(self, eip, project_id, graph_name, token):
        self.eip = eip
        self.project_id = project_id
        self.graph_name = graph_name
        self.headers = {'X-Auth-Token': token, 'Content-Type': 'application/json'}

    def build_vertex_index(self):
        url = ('http://{}/ges/v1.0/{}/graphs/{}/indices').format(self.eip,  self.project_id, self.graph_name)
        body = json.dumps({
            "indexName": "cypher_vertex_index",
            "indexType": "GlobalCompositeVertexIndex",
            "hasLabel": "true",
            "indexProperty": []
        })
        r = requests.post(url=url, data=body, headers=self.headers)
        return r.json()['jobId']

    def build_edge_index(self):
        url = ('http://{}/ges/v1.0/{}/graphs/{}/indices').format(self.eip, self.project_id, self.graph_name)
        body = json.dumps({
            "indexName": "cypher_edge_index",
            "indexType": "GlobalCompositeEdgeIndex",
            "hasLabel": "true",
            "indexProperty": []
        })
        r = requests.post(url=url, data=body, headers=self.headers)
        return r.json()['jobId']

    def get_job(self, job_id):
        url = ('http://{}/ges/v1.0/{}/graphs/{}/jobs/{}/status').format(self.eip, self.project_id, self.graph_name, job_id)
        r = requests.get(url=url, headers=self.headers)
        output = r.json()
        return output

    def summary(self):
        url = ('http://{}/ges/v1.0/{}/graphs/{}/summary?label_details=true').format(self.eip, self.project_id, self.graph_name)
        r = requests.get(url=url, headers=self.headers)
        output = r.json()['data']
        return output

    def cypher_query(self, statement):
        url = ('http://{}/ges/v1.0/{}/graphs/{}/action?action_id=execute-cypher-query').format(self.eip, self.project_id, self.graph_name)
        body = json.dumps({
            "statements": [
                {
                    "statement": statement,
                    "parameters": {},
                    "resultDataContents": [
                        "row"
                    ],
                    "includeStats": False
                }
            ]
        })
        r = requests.post(url=url, data=body, headers=self.headers)
        output = r.json()['results']
        return output
    
    def format_cypher_result(self, json_obj):
        for x in json_obj:
            columns = x["columns"]
            data = x["data"]
            rows = []
            for y in data:
                rows.append(y["row"])
            return pd.DataFrame(rows, columns=columns)

创建好图之后，就可以对其进行查询和分析了。

# 需填入参数
iam_url = ''
user_name = ''
password = ''
domain_name = ''
project_name = ''
eip = ''
project_id = ''
graph_name = ''
config = config(iam_url, user_name, password, domain_name, project_name, eip, project_id, graph_name)
ges_util = GESFunc(config.eip, config.project_id, config.graph_name, config.token)

/home/ma-user/anaconda3/lib/python3.6/site-packages/urllib3/connectionpool.py:858: InsecureRequestWarning: Unverified HTTPS request is being made. Adding certificate verification is strongly advised. See: https://urllib3.readthedocs.io/en/latest/advanced-usage.html#ssl-warnings

  InsecureRequestWarning)

GES 支持 cypher 查询语言，后续的查询示例使用的是 cypher 查询语言。在使用 cypher 查询之前，我们先创建点索引和边索引。

print('开始创建点索引:')
job_id = ges_util.build_vertex_index()
job_result = ges_util.get_job(job_id)
if 'errorCode' not in job_result:
    for i in range(100):
        if job_result['status'] == 'success':
            break
        else:
            time.sleep(1)
            job_result = ges_util.get_job(job_id)
print('点索引创建完成')

开始创建点索引:

点索引创建完成

print('开始创建边索引:')
job_id = ges_util.build_edge_index()
job_result = ges_util.get_job(job_id)
if 'errorCode' not in job_result:
    for i in range(100):
        if job_result['status'] == 'success':
            break
        else:
            time.sleep(1)
            job_result = ges_util.get_job(job_id)
print('边索引创建完成')      

开始创建边索引:

边索引创建完成

查询演示

下面，我们可以书写并运行标准的 cypher 语言，对这份图数据进行探索了。以下是部分图数据的可视化展示。

查询这份图数据的统计信息：

print('统计信息:')
result = ges_util.summary()
res_str = json.dumps(result, indent=4)
print(res_str)

统计信息:

{

    "vertexNum": 2449792,

    "labelDetails": {

        "labelInVertex": {

            "paper": 712464,

            "author": 1737328

        },

        "labelInEdge": {

            "reference": 670627,

            "write": 4434198

        }

    },

    "edgeNum": 5104825

}

列举若干 paper 的信息：

print('查询 papers:')
statement = "match (n:paper) return n.title limit 5"
paper_result = ges_util.cypher_query(statement)
format_result = ges_util.format_cypher_result(paper_result)
format_result

查询 papers:

.dataframe tbody tr th {
    vertical-align: top;
}

.dataframe thead th {
    text-align: right;
}

查询某 paper 的 authors:

print('查询某 paper 的 authors:')
paper = paper_result[0]['data'][1]['row'][0]
statement = "match (n:paper)<--(m:author) WHERE n.title = '" + paper + "' return id(m)"
result = ges_util.cypher_query(statement)
format_result = ges_util.format_cypher_result(result)
format_result

查询某 paper 的 authors:

.dataframe tbody tr th {
    vertical-align: top;
}

.dataframe thead th {
    text-align: right;
}

查询某 paper 被引用的次数：

print('查询某 paper 被引用次数:')
paper = paper_result[0]['data'][2]['row'][0]
statement = "match (n:paper)<--(m:paper) WHERE n.title = '" + paper + "' return count(m)"
result = ges_util.cypher_query(statement)
format_result = ges_util.format_cypher_result(result)
format_result

查询某 paper 被引用次数:

.dataframe tbody tr th {
    vertical-align: top;
}

.dataframe thead th {
    text-align: right;
}

查询某 paper 与哪些 paper 被联合引用及其次数:

print('查询某 paper 与哪些 paper 被联合引用及其次数:')
paper = paper_result[0]['data'][3]['row'][0]
statement = "match (n:paper)<--(p:paper)-->(m:paper) WHERE n.title = '" + paper + "' return m.title, count(*) as p order by p desc limit 5"
result = ges_util.cypher_query(statement)
format_result = ges_util.format_cypher_result(result)
format_result

查询某 paper 与哪些 paper 被联合引用及其次数:

.dataframe tbody tr th {
    vertical-align: top;
}

.dataframe thead th {
    text-align: right;
}

查询标题带关键字"Virus"的 paper 数量:

print('查询标题带关键字"Virus"的 paper 数量:')
statement = "MATCH (p:paper) WHERE p.title IS NOT NULL AND p.title CONTAINS('Virus') RETURN count(p)"
result = ges_util.cypher_query(statement)
format_result = ges_util.format_cypher_result(result)
format_result

查询标题带关键字"Virus"的 paper 数量:

.dataframe tbody tr th {
    vertical-align: top;
}

.dataframe thead th {
    text-align: right;
}

查询标题带关键字"Virus"的 paper，并按发表日期排序:

print('查询标题带关键字"Virus"的 paper，并按发表日期排序:')
statement = "MATCH (p:paper) WHERE p.publish_time IS NOT NULL AND p.title IS NOT NULL AND p.title " \
            "CONTAINS('Virus') RETURN p.title, p.publish_time ORDER BY p.publish_time DESC LIMIT 5"
result = ges_util.cypher_query(statement)
format_result = ges_util.format_cypher_result(result)
format_result

查询标题带关键字"Virus"的 paper，并按发表日期排序:

.dataframe tbody tr th {
    vertical-align: top;
}

.dataframe thead th {
    text-align: right;
}

查询 paper 被引用的次数并以此排序:

print('查询 paper 被引用的次数并以此排序:')
statement = "match (m:paper)<--(p:paper) with m, count(p) as citedNum return m.title, citedNum order by citedNum desc limit 5"
result = ges_util.cypher_query(statement)
format_result = ges_util.format_cypher_result(result)
format_result

查询 paper 被引用的次数并以此排序:

.dataframe tbody tr th {
    vertical-align: top;
}

.dataframe thead th {
    text-align: right;
}

列举若干 author 的信息:

print('查询 authors:')
statement = "match (n:author) return id(n) limit 5"
author_result = ges_util.cypher_query(statement)
format_result = ges_util.format_cypher_result(author_result)
format_result

查询 authors:

.dataframe tbody tr th {
    vertical-align: top;
}

.dataframe thead th {
    text-align: right;
}

查询某 author 的合作者及合作次数:

print('查询某 author 的合作者及合作次数:')
author = author_result[0]['data'][0]['row'][0]
statement = "match (n:author)-->(p:paper)<--(m:author) WHERE id(n) = '" + author + "' return id(m), count(*) as p order by p desc limit 5"
result = ges_util.cypher_query(statement)
format_result = ges_util.format_cypher_result(result)
format_result

查询某 author 的合作者及合作次数:

.dataframe tbody tr th {
    vertical-align: top;
}

.dataframe thead th {
    text-align: right;
}

查询某 author 论文被引用次数:

print('查询某 author 论文被引用次数:')
author = author_result[0]['data'][0]['row'][0]
statement = "match (m:author)-->(p:paper)<--(n:paper) where id(m) = '" + author + "' return count(n)"
result = ges_util.cypher_query(statement)
format_result = ges_util.format_cypher_result(result)
format_result

查询某 author 论文被引用次数:

.dataframe tbody tr th {
    vertical-align: top;
}

.dataframe thead th {
    text-align: right;
}

查询论文互相引用情况:

print('查询论文互相引用情况:')
statement = "match (m:paper)-->(p:paper)-->(n:paper) where id(m) <> id(p) and id(m) = id(n) return m,p limit 10"
result = ges_util.cypher_query(statement)
format_result = ges_util.format_cypher_result(result)
format_result

查询论文互相引用情况:

.dataframe tbody tr th {
    vertical-align: top;
}

.dataframe thead th {
    text-align: right;
}