<p><img src="https://www-cdn.nebula-graph.com.cn/nebula-blog/Release-note-v1.png" alt="Release-Note" /></p>
<p>NebulaGraph 发布 1.0.1 版。相比 1.0.0 版本，该版本针对系统稳定性、性能上做了很多优化和改进。</p>
<h2 id="post-new-features">New Features</h2>
<ul>
<li>路径查询 <code>FIND PATH</code> 中 vid 支持 <code>uuid()</code>  https://github.com/vesoft-inc/nebula/pull/2209</li>
<li>增加了 disable rocksdb block cache 的功能，通过设置 <code>storaged.conf</code> 的 <code>rocksdb_block_cache</code> 为任意非正数来 disable block cache.  https://github.com/vesoft-inc/nebula/pull/2177</li>
<li>避免了多线程对 meta client 的 <code>localCacheLock_</code> 的lock contention 等待事件，提高了性能 https://github.com/vesoft-inc/nebula/pull/2165</li>
<li>底层存储支持多种压缩算法, 支持对 rocksdb 每个level 指定压缩策略  https://github.com/vesoft-inc/nebula/pull/2179</li>
</ul>
<h2 id="post-bug-fix">bug-fix</h2>
<ul>
<li>修复手动 Compaction 时 <code>CompactionFilter</code> 失效的问题 https://github.com/vesoft-inc/nebula/pull/2184</li>
<li>修复某些情况下，<code>Leader</code> 错误发送 <code>Snapshot</code> https://github.com/vesoft-inc/nebula/pull/2150</li>
<li>修复增加属性后，<code>UPSERT/UPDATE SET xxx</code> 老数据的新字段报错的问题 https://github.com/vesoft-inc/nebula/pull/2186</li>
<li>修复 <code>GO m ... n</code> 当 m 为 0 可能导致的问题 https://github.com/vesoft-inc/nebula/pull/2202</li>
<li>修复 pipe 操作结合 GO n STEPS 查询时，某些情况下可能返回数据不全的问题 https://github.com/vesoft-inc/nebula/pull/2203</li>
</ul>
<p>喜欢本次更新内容？来来来，给我们的 <a href="https://github.com/vesoft-inc/nebula-graph">GitHub</a> 点个 star 表鼓励啦~~ 🙇‍♂️🙇‍♀️ [手动跪谢]</p>
<p>交流图数据库技术？加入 Nebula 交流群请先<a href="https://wj.qq.com/s2/8321168/8e2f/">填写下你的 Nebula 名片</a>，Nebula 小助手会拉你进群~~</p>
<h2 id="post-推荐阅读">推荐阅读</h2>
<ul>
<li><a href="https://nebula-graph.com.cn/posts/nebula-graph-1.0-release-note/">NebulaGraph 1.0 Release Note</a></li>
</ul>

NebulaGraph 发布 1.0.1 版。相比 1.0.0 版本，该版本针对系统稳定性、性能上做了很多优化和改进。

用户案例

周边工具

特性讲解

开发日志

技术分享

产品实践

社区动态

行业科普

新闻资讯

发版说明

安装部署

解决方案

架构系列

源码解读

NUC2022

性能报告

NebulaGraph v1.0.1 Release Note

Nebula 架构剖析系列（零）图数据库的整体架构设计

NebulaGraph发布企业版 v3.4.0，深入探索数据价值

美团：主流图数据库厂商选型报告 NebulaGraph vs Dgraph vs JanusGraph

中国移动：NebulaGraph 图数据库在金融风控中的落地应用

如何用 NebulaGraph 图技术搞定 7 个典型社交网络应用？

![LDBC](https://nebula-website-cn.oss-cn-hangzhou.aliyuncs.com/nebula-website-5.0/share/2024-04-23-1.jpg)

LDBC（Linked Data Benchmark Council）Social Network Benchmark，简称 LDBC SNB，是一种针对社交网络场景的评估图数据库性能的基准测试。

## **LDBC 简介 **

除了 Social Network Benchmark，LDBC 旗下目前还有其他几种基准测试：Graphalytics Benchmark，Financial Benchmark 和 Semantic Publish Benchmark，分别针对图分析、金融和 RDF 的场景。Social Network Benchmark 是 LDBC 最早的提出的基准测试，已经成为国内外最主流的图数据库基准测试，在国内很多图数据库招标也会将 LDBC SNB 作为性能测试的一项。但需要说明的是，LDBC 本身作为一个非盈利组织，只提供官方审计。**不同图数据库可能受到运行环境以及基准测试的相关参数影响，因此测试结果的横向对比没有任何意义。**

LDBC SNB 主要包括三个主要部分：

1**.Data Generator**：这是一个数据生成工具，用于生成具有社交网络特性的大规模复杂数据。这些数据包括人、帖子、评论、地理位置、组织和其他一些社交网络的典型实体和关系。

2.**Interactive Workload**：主要针对 OLTP，模拟了用户在社交网络上的日常活动，例如发布帖子、添加好友、点赞等。读请求以查询以一到两跳为主，同时可能会伴随一些写请求。

3.**Business Intelligence Workload**：主要针对 OLAP，模拟了对社交网络数据进行深入分析，以全图查询为主。例如分析用户的社交行为、社区的形成和演变，以及其他一些需要复杂分析和大量数据处理的任务。

LDBC SNB 的论文里还提到了一个 SNB Algorithms，顾名思义主要是跑图算法的，如 PageRank、社区发现、广度搜索等。但论文是 2015 年发表的，当时描述这个场景还在起草中，目前已经将这部分移到了 Graphalytics Benchmark。

https://discuss.nebula-graph.com.cn/t/topic/15080

此外，想要运行 LDBC SNB 测试，还需要一个官方提供的 Driver。不同的数据库需要基于 Driver 的接口实现相应的 Connector，用来连接 Driver 和数据库。之后 Driver 会根据 Benchmark 的相关参数生成 Workload（这里可以理解为一系列的查询语句），并驱动待测数据库执行这些查询语句，最终得到性能测试结果。

整个 LDBC SNB 基准测试的流程如下，主要分成：**准备阶段、基准测试、结果输出**这三个阶段。

准备阶段主要执行数据生成，包括初次导入的全量数据，以及后续实时更新的数据。此外在官方审计中，还需要在 SF10 Dataset（在本文文末介绍了何为 SF）上进行 Validation，因此这一阶段也会生成用于校验的数据。

基准测试阶段会先在 SF10 Dataset 上进行 Validation，之后会在 SF30 或者 SF100 Dataset 进行性能测试。Validation 的过程就是在数据导入之后，由 Driver 根据之前准备阶段的一系列 query 和期望结果，对数据库的查询结果进行校验，以确保数据库的查询结果正确。Validation 的这个过程没有时间要求。而之后的性能测试分为导入、预热、性能测试，数据库可以有 30 分钟的预热时间，而在性能测试至少要持续两个小时，最终将测试结果汇总并输出。

![Validation](https://nebula-website-cn.oss-cn-hangzhou.aliyuncs.com/nebula-website-5.0/share/2024-04-23-2.jpg)

由于篇幅限制，我们这一系列重点介绍 SNB Interactive Workload 相关内容。这一篇，我们主要会结合论文，介绍 SNB 的 Schema 以及数据生成，也就是准备阶段。

## **LDBC SNB Schema 生成**  

说明：为了和 SNB 中的数据命名统一，本文相关名称我会用英文，所以读起来可能会有些怪怪的。为了降低理解成本，每个英文单词首次出现后面会跟随对应的中文注释讲解。

SNB 的数据主要是模拟了一个类似 Facebook 的社交网络。其中数据都是围绕 Person（人）构建而来，Person 之间会构成 Friendship（情谊）网络。每个 Person 可能会有若干 Forum（特定讨论区），Person 可以在 Forum 中下面发送若干 Post（帖子），其他 Person 可能会 likes（点赞）其中一些 Message（消息）。

以上这些元素的数据量主要会受 Person 和时间的影响：

- 有更多朋友的人会发送更多的评论或点赞

- 时间越长，会结交更多的朋友，评论或点赞数量也会上升

还有一部分数据不会随 Person 数量而变化，主要包括一些 Organization（组织，这里主要是学校）以及 Place（地方，这里主要是居住城市、国家等地理信息）。这部分数据会在数据生成时起一些作用，比如在同一时期在同一个学校上学的人更有可能称为朋友。

SNB 的完整 Schema 如下图所示：

![Schema](https://nebula-website-cn.oss-cn-hangzhou.aliyuncs.com/nebula-website-5.0/share/2024-04-23-3.jpg)

大多数图数据库在进行测试时候，会将实体建模为点，而不同关系会建模为边。但这只是一个惯例，**SNB 的数据建模和实际数据库中的 Schema 可以不同，只要数据库能够完成相应 Workload 的查询即可**。

### **LDBC SNB 数据生成  **

SNB 的一个重要部分是 Data Generator（下文称为 DataGen），用来生成满足上面 Schema 的数据。Generator 生成的数据由以下三个参数决定：

- Person 的个数

- 模拟多少年的数据

- 从哪一年开始模拟

根据官方文档，DataGen 生成的数据有以下性质：

- 现实性：生成数据模拟了一个真实的社交网络。一方面，生成数据中的属性、基数、数据相关性和分布经过精心设置，从而能够模拟 Facebook 等真实社交网络。另一方面，其原始数据来自于 DBpedia，保证数据中的属性值真实且相关。

- 可扩展性：针对不同规模和预算的系统，DataGen 能够生成不同大小的数据集（GB 到 TB 级），此外 DataGen 可以在单机，或者是一个集群中完成数据生成。

- 确定性：无论用来生成数据的机器数量多少、机器配置是高还是低，DataGen 生成的数据都是确定且相同的。这一重要功能确保了任意一个数据系统都能使用相同的数据集，保证不同系统环境之间的测评比较公平且基准测试结果可重复。

- 易用性：DataGen 被设计得尽可能易于使用。

整个数据生成的流程图如下所示，我们会分解为几部分介绍：

![DataGen](https://nebula-website-cn.oss-cn-hangzhou.aliyuncs.com/nebula-website-5.0/share/2024-04-23-4.jpg)

### 生成属性分布

第一步是初始化。DataGen 使用的原始数据来自于 DBpedia，针对每一个属性，DataGen 会根据以下方面决定属性的分布：

- 有多少种可能的属性值

- 每一种属性值出现的概率

最终将属性的分布情况作为资源文件以及 DataGen 的参数保存下来。

### 生成 Person 和 Friendship

前面也提到过，SNB 的 Schema 的核心是 Person，这也体现在数据生成过程中。接下来 DataGen 就会生成所有 Person，以及 Person 中一部分后续操作所需要的信息，比如每个 Person 有多少 Friendship（这个值非常重要，其分布满足 Power law（幂定律）），Person 所就读的大学，Person 所就职的公司等。

接下来，DataGen 会创建每个 Person 的 Friendship 关系（即流程图中的 knows）。和真实社交网络一样，有相同兴趣或者行为的人，很有可能会连接在一起。为了模拟这样的社交网络，SNB 在生成 Friendship 时会考虑以下三个维度：

- Person 所就读的大学，就读时间，以及大学所在城市

- Person 的兴趣

- 每个 Person 会生成一个随机值，随机值越相近代表其越类似（这是为了模拟不是所有朋友都是通过大学和兴趣结交的）

三个维度分别占每个 Person 的 Friendship 关系权重的 45%，45% 和 10%，也就将 Person 之间建边的过程分成了三个子步骤。

DataGen 会依次根据三个维度将所有 Person 进行排序（每次只按一个维度进行排序），然后将排序过后的 Person 切分为不相交的多个部分，分发给不同 Worker 进程。即便是切分之后，每个 Worker 线程负责的 Person 可能也可能超过内存大小。因此，Worker 线程会维护一个滑动窗口，滑动窗口内的 Person 之间建立 Friendship 关系的概率满足几何分布。

如下图所示：

![Friendship](https://nebula-website-cn.oss-cn-hangzhou.aliyuncs.com/nebula-website-5.0/share/2024-04-23-5.jpg)

假设现在根据就读大学这个维度进行了排序，得到了一个 Person 有序序列。之后 Worker 就会维护一个滑动窗口，每次为滑动窗口最左侧的人生成 Friendship 关系（上图当前是 P2），滑动窗口内的其他人和窗口第一个人建立 Friendship 的比例满足几何分布。

直到滑动窗口的第一个人建立了足够多的 Friendship 之后，滑动窗口的起点会移到下一个人。

这里没有深究滑动窗口的大小、几何分布的参数甚至是随机生成器的参数，不知道在出现滑动窗口内无法生成足够多 Friendship 关系时，DataGen 如何处理。

将三个维度都经过排序、分发、按滑动窗口建边之后，DataGen 就进入了下一阶段。

### 生成社交活动

生成完 Person 和 Friendship 之后，DataGen 就开始生成每个 Person 的社交活动，包括 Forum，Post 和 Comment。这部分数据也有一些相关性存在：

1.有越多 Friendship 的 Person 在社交网络上会越活跃

2.每个 Person 更可能在自己感兴趣或者就读大学相关的 Forum 进行 Post 或者 Comment

3.社交活动和时间是有相关性的，比如接近世界杯，足球相关的讨论就会激增

### 最终输出

经过以上步骤之后，DataGen 完成了数据生成，模拟的社交网络图会分成两部分进行输出：

- Dataset：90% 的数据用于初始导入

- Update Streams：10% 的数据用于后续实时更新

除此之外，还会生成后续 Workload 中请求的参数（主要是起点）。关于参数生成我们会在下一篇详细解释，这里简单描述一下 SNB 的读请求 Workload。Interactive Workload 主要的查询希望在一秒以内得到查询结果，所有读 query 都是从图中的一个点出发，获取很小一部分的子图信息。另外，因不同起点的出入度不同，基本上也就决定了这次读请求会访问的数据量。

为了测试不同系统和场景，SNB 定义了**比例因子（Scale Factor，即所谓的 SF）用来控制最终生成的数据量大小**。比如，SF1 原始数据大小为 1 GB，同理 SF0.1 和 SF300 的大小为 100 MB 和 300 GB。不同比例因子的各个类型的点边数据量如下表所示：

![Scale Factor](https://nebula-website-cn.oss-cn-hangzhou.aliyuncs.com/nebula-website-5.0/share/2024-04-23-6.jpg)

最终生成的 Dataset 分为两大类：Static 和 Dynamic，格式都是 CSV。根据 DataGen 配置的线程数量大小，最终生成的数据也会分为多个分片。Static 包含 Organization、Place、Tag 等，都是基于 DBpedia 生成的静态数据，其数量不会随着比例因子变化而变化。换而言之，这部分数据与 Person 的个数无关。而 Dynamic 部分主要包括 Person、knows（即前面数据生成部分描述的 Friendship）、Forum、Post 和 Comment 等。

而 Update Streams 中包含了所有更新的操作，主要就是模拟实时注册新用户、评论、点赞、加好友等等行为。

## 参考文献  

到这里准备阶段大概就介绍完了，在准备阶段最终生成的请求参数部分我们会在下一篇讲述 Workload 时再展开。

- ldbc-snb-interactive-sigmod-2015.pdf (ldbcouncil.org)

- The LDBC Social Network Benchmark Specification


## 关于 NebulaGraph  

NebulaGraph 是一款开源的分布式图数据库，自 2019 年开源以来，先后被美团、京东、360 数科、快手、众安金融等多家企业采用，应用在智能推荐、金融风控、数据治理、知识图谱等等应用场景。GitHub 地址：https://github.com/vesoft-inc/nebula

DataGen 会依次根据三个维度将所有 Person 进行排序（每次只按一个维度进行排序），然后将排序过后的 Person 切分为不相交的多个部分，分发给不同 Worker 进程。

图数据库基准测试 LDBC SNB 系列讲解：数据测评公平、结果可重复的秘密

![Workload](https://nebula-website-cn.oss-cn-hangzhou.aliyuncs.com/nebula-website-5.0/share/2024-04-17-1.webp.jpg)

继上文的 LDBC 数据生成🔗之后，在这一篇我们主要介绍 SNB Workload 里的各个 Query 是如何生成的，以及在性能测试中 Driver 是如何执行 Workload。

**Workload**

在基准测试的准备阶段之后，总共生成了三类数据：
- **Dataset**：用于初始导入的数据
- **Update Streams**：用于 Driver 在性能测试阶段实时更新
- **Parameters**：用于 Driver 在性能测试阶段读请求的参数

### 写请求 Write Query

由于 SNB 生成的数据非常复杂，因此在性能测试阶段不可能通过 SNB Driver 实时生成更新请求，因此**实时更新的数据实际是由 DataGen 来生成的**。我们在上一篇也提到过，使用 LDBC 时 DataGen 需要指定生成多长时间的数据，以及社交活动是从哪一年开始。DataGen 会根据时间戳将所有数据进行切分，将 90% 数据作为 Dataset，剩下的 10% 数据作为 Update Streams 用于后续更新的写请求。

对于所有写请求，DataGen 都会将这次写请求（Update Query）调度执行的时间、相关更新的参数生成到一个 JSON 中。（至于为什么需要指定 Query 调度时间，我们后续会解释）

### 读请求 Read Query

SNB 的读请求分为两大类：14 个 Complex read-only queries 以及 7 个 Short read-only queries。

Complex 读请求都是以一个给定 Person 出发，查询一跳或者两跳子图内的相关信息。比如 Interactive Complex 9（简称 IC9）的查询 Pattern 如下：给定一个起始的人，查询这个人的朋友或者朋友的朋友创建的最近 20 条消息。如下图所示：

![Read Query](https://nebula-website-cn.oss-cn-hangzhou.aliyuncs.com/nebula-website-5.0/share/2024-04-17-2.webp.jpg)
图注：Interactive / complex / 9

Short 读请求分为两大类：一类是给定一个人，查询这个人的姓名、年龄等信息，或者是这个人的朋友以及这些朋友发送的消息。另一类则是给定一个消息，返回这条消息是的发送者以及发送者的相关信息。类似下图这样：

![Short ](https://nebula-website-cn.oss-cn-hangzhou.aliyuncs.com/nebula-website-5.0/share/204-04-17-3.webp.jpg)
图注：Interactive / short / 1

完整的请求描述可以参见**《The LDBC Social Network Benchmark Specification》** 的 6.1 和 6.2 章节，这里就不展开了。

### 请求参数

除了 Dataset 和 Update Streams 之外，DataGen 还会生成不同读请求的参数（比如上面 IC9 的起点 PersonId）。为了保证性能测试的结果能够稳定的衡量数据库的性能，SNB 保证生成读请求的参数满足以下性质：

1.查询时间有上限：Complex Read 和 Short Read 请求所涉及的数据会因为起点不同而不同，但从宏观上看，其查询的子图都是有限的，因此其查询时间也会有一个合理的范围。

2.查询执行时间满足稳定的分布：同一个读请求的各个起点不同，查询数据量会不一样，但最终查询时间分布是稳定的。

3.逻辑计划相同：给定一个查询，无论起点如何不同，最终经过优化器优化之后生成的逻辑查询计划是相同的。这条性质保证每一个查询，能够稳定测试到图数据库中的一些技术难点（比如大量 JOIN、基数估计等）。

其中前两条还比较好理解，但最后一条可能做到吗？我们都知道一个优秀的优化器会根据查询参数决定实际执行时的 JOIN 顺序。即便对于相同的查询 Pattern，由于查询参数不同，最终的执行计划也会千变万化。我们看下** DataGen 是如何筛选参数，使同一个查询最终采用相同的查询计划。**

首先，我们都知道查询数据量和查询时间是强相关的。对于每一个查询 Pattern，LDBC 在设计时都会有一个理想执行计划，在这个理想执行计划中，其数据量是可以通过起点预先计算出来的。

比如 Complex 2：

![Complex 2](https://nebula-website-cn.oss-cn-hangzhou.aliyuncs.com/nebula-website-5.0/share/2024-04-17-4.webp.jpg)

获取给定某个人的朋友发送的最近 20 个消息，其理想查询计划如下所示。

![Parameter-Count table](https://nebula-website-cn.oss-cn-hangzhou.aliyuncs.com/nebula-website-5.0/share/2024-04-17-5.webp.jpg)

对于这个理想执行计划，我们可以预处理得到每个起点的 JOIN 的中间结果行数。比如对于 PersonID=1542 的起点，两次 JOIN 的中间结果的数量分别为 60 和 99。最终，对于每个 Query，会生成一个起点和中间 JOIN 结果行数的表，称为 Parameter-Count table。

**这个 Parameter-Count table 是由 DataGen 在生成数据过程中一起生成的。**有了这张表，挑选参数的问题就变成了在这张表中挑选若干最相近的行。在这一步中采用的方式是贪心的启发式搜索：首先根据 Join1 中间结果数量，挑选最相近的若干行（Join1 中值为 60 的几行），然后在这些行中再根据 Join2 的中间结果数量进行挑选（得到 Join2 值为 99、102、103 的几行），最终筛选出来的 PersonID 为1542、1673、7511。

这里存疑的是，按理来说应该先确定需要挑选参数的个数 k，然后再在 Parameter-Count table 进行筛选，否则按论文所说方法，可能会导致最终挑选的行数小于 k。这里没有深究，但整体思路是合理的。

通过上述步骤，对于每一个 Query 类型，都会挑选出来若干个最相近的参数，从而尽可能使得查询计划相同。不过还剩一个问题，如何确定每个查询需要多少个参数，也就是不同查询的数量呢？

### Query Mix

首先，LDBC SNB Interactive Workload 期望数据库在性能测试过程中，**Update Query、Complex Query 和 Short Query **的执行时间比例依次是 **10%、50% 和 40%**。这个比例一方面是为了模拟真实的社交网络负载，一般来说读请求会远多于写请求，另外这个负载也要能够真实反映数据库的性能。

尽管在上一步中，DataGen 保证了同一个查询中的不同参数的执行时间相近。但不同查询的数据量和执行数据有巨大的差别。因此为了能够避免某一个类型的查询对最终测试结果起决定性作用，不同类型查询的数量需要精心的设计。

**SNB Interactive Workload 主要认为 Complex Read Query 应该主导是性能测试分数**。Complex Read Query 的数量是由 Update Query 的数量决定的，如下表所示：

![Complex Read Query](https://nebula-website-cn.oss-cn-hangzhou.aliyuncs.com/nebula-website-5.0/share/2024-04-17-6.webp.jpg)

比如在 SF100 中每执行 527 个 Update Query 之后应该执行一次 Complex Read Query 9。前面提过，每个 Update Query 的调度时间由 DataGen 已经决定了，因此每一个 Complex Read 的调度时间也就确定了。**这里可以暂时将调度时间理解为相对于性能测试开始时间的一个相对时间，这个相对时间实际还会受到 SNB Driver 相关参数的影响。**之所以是调度时间而不是执行时间的原因在于待测数据库可能已经满负载或者没有响应，导致某个 Query 可能已经到了调度时间，但无法及时执行 Query。

这里就可以理解为什么 SNB 号称其测试结果可重复，关键之处就是 DataGen 生成的数据是确定性的。一旦 Update Query 调度时间确定，也就确定了 Complex Read Query 的调度时间，进而确保其性能测试结果可重复。了解这一点对于后续理解性能测试的结果以及 SNB Driver 的相关参数是至关重要的。

而 Short Query 会被穿插在每一个 Complex Query 之后。前面提到过 Short Query 分为两大类，查询人的相关信息以及查询消息的相关信息。准确来说，不同类型的 Complex Query 可能会返回不同信息，根据 Complex Query 的查询结果，后续会从 Short Query 中挑选一类，并把 Complex Read 的结果（比如 Person 或者 Message 的标识符）作为 Short Query 的参数进行查询。

一个 Short Query 被调度之后，后续有可能会调度另一个 Short Query，每次概率逐渐降低。即一个 Complex Query 之后，可能调度一连串 Short Query（论文中称为 Short Read Random Walk）。不同的 Query 后面可能触发的 Short Query 类型不同，如下表所示，比如 IC10 之后就只能触发 IS1、IS2 和 IS3。
![Short Query](https://nebula-website-cn.oss-cn-hangzhou.aliyuncs.com/nebula-website-5.0/share/2024-04-17-7.webp.jpg)

图注：IC 为 Interactive Complex query 的缩写，IS 为 Interactive Short query 的缩写

和 Complex Query 不同的时，Short Query 的调度时间是由 Driver 根据它所依赖的 Complex Read 的完成时间所动态决定的。但触发多少个 Short Query 就是由上面说的概率决定，而所有性能测试过程中使用的随机生成器是相同的，综上所述整个基准测试过程中所有 Query 的比例和顺序是确定的。

**Driver **

Driver 依赖 DataGen 生成的 Dataset、Update Streams 和 Parameters，它主要能完成两个任务：

- Validate：校验数据库的正确性
- Benchmark：验证数据库的性能

Validate 比较简单，根据 Update Streams 和 Parameters，生成若干用于校验的参数，**为了验证数据库的结果正确性，Validate 依赖于确定的顺序执行查询，因此只能单线程执行。**Driver 最终会在数据库输出是否通过验证的结果。

Benchmark 则是根据 Dataset、Update Streams 和 Parameters 以及相关配置，生成确定的 Workload（即一个 Operation 序列，其中包含 Query 和对应调度时间）。根据调度时间，按顺序发送给待测数据库，同时保存其调度时间、开始执行时间和结束执行时间，最终输出性能报告。

Driver 主要由 Runtime、Workload、Database Connector、Report 这几部分组成。

### Runtime & Workload

无论最终性能测试在什么环境、待测图数据库是什么，只要 Driver 的输入（数据和配置）相同，其生成的 Workload 就是相同的。为了得到最高的吞吐量，一般会把 Workload 拆成多个 Stream。对于图来说，多个操作是有前后依赖关系的，比如好友存在才能添加好友关系。因此，Driver 支持由多个线程来执行这个序列（由 thread_count 参数控制），但也需要保证即便有多个线程来执行，操作之前的依赖关系没有被破坏，仍然满足 DataGen 生成的调度时间关系。这部分原理就不展开介绍，感兴趣的可以阅读论文中的相关部分。

Workload 中的每个 Operation 包括以下信息：

- **Query 类型**

- **Query 参数**

- **调度时间**

- **期望结果（只在 Validate 中使用）**

前面提到过所有 Update Query 和 Complex Query 的调度时间都是 DataGen 预先确定好的，Driver 会根据调度时间通过 Database Connector，把 Query 发送给数据库执行。显然，如果这些调度时间都是一个固定值，我们是无法获知数据库的真实性能的。

在 Driver 中有一个参数 time_compression_ratio（默认为 1），可以在时间维度上拉伸或者压缩整个 Workload，比如 time_compression_ratio 为 2.0 代表这个 benchmark 的运行持续时间会变成默认的两倍，也就是将 Query 之间的间隔时间变为原来的两倍。同理，time_compression_ratio 为 0.1 代表 Driver 发送 Query 的速度是默认情况下的 10 倍。简而言之：
- time_compression_ratio > 1，比默认慢
- time_compression_ratio < 1，比默认快

注意，修改 time_compression_ratio 不会影响 Query 序列中各个类型 Query 的比例、Query 的参数和执行顺序。

### Database Connector

为了将数据库对接至 SNB，厂商需要实现对应的 Database Connector。其主要任务就是对接目标数据库：

1.将 Driver 框架的 Query 类型和参数，转换为目标图数据库的 Query
2.将 Query 通过 Database Connector 发送给目标图数据库
这部分都只需要实现 Driver 的若干接口即可，可以参考：**https://github.com/ldbc/ldbc_snb_interactive_v1_driver **以及相关集成文档。

### Driver Configuration

Benchmark 的主要目的就是在保证大于 95% 的及时率前提下，得到最优的 time_compression_ratio（越小代表性能越高）。Operation 如果在期望调度时间过后的 1 秒之内被 Driver 调度，就认为是及时的：

actual_start_time − scheduled_start_time < 1 second

此外，一次有效的 Benchmark 需要至少需要测试两个小时（预热的半个小时期间，性能是不计入在结果之内）。

![Benchmark](https://nebula-website-cn.oss-cn-hangzhou.aliyuncs.com/nebula-website-5.0/share/2024-04-17-7.webp.jpg)

为了运行一次有效的 Benchmark，Driver 有几个重要的参数需要调整：

- time_compression_ratio：调整 Workload 的执行速度，准确说是期望执行速度
- operation_count：Driver 总共需要生成多少个 Operation，配合 - time_compression_ratio 修改，以保证 Benchmark 的时长
- thread_count：执行 Operation 的线程池大小，根据待测数据库的性能，可能也需要调整并发执行 Query 的数量

### Report

最终 Driver 会生成如下性能指标：
- Start Time：开始时间，第一个 Operation 开始时间
- Finish Time：完成时间，最后一个 Operation 完成时间
- Total Run Time：运行时间，Finish Time 和 Start Time 的差值
- Total Operation Count：Operation 总数，总共执行了多少个 Operation

系统的 QPS 可以通过 Total Operation Count / Total Run Time 得到。

对于每个类型的 Query，Driver 也会输出每个 Query 的总数，执行时间的平均值、中位数、P95、P99等、以及调度时间的延迟分布等。

![Query](https://nebula-website-cn.oss-cn-hangzhou.aliyuncs.com/nebula-website-5.0/share/2024-04-17-9.webp.jpg)

图注：LDBC 报告示意图


**参考文献**

有关 LDBC SNB Interactive 的介绍基本就到这了，后续可能还会总结一下这个基准测试的一些设计思想。另外我们自己系统在测试过程中，也出现了一些因为对 SNB 本身不够理解而当时看来十分诡异的现象，这也是花时间总结这几篇文章的初衷。

- ldbc-snb-interactive-sigmod-2015.pdf (ldbcouncil.org)
- The LDBC Social Network Benchmark Specification
- GitHub - ldbc/ldbc_snb_interactive_v1_driver: Driver for the LDBC SNB Interactive workload

## 关于 NebulaGraph  

NebulaGraph 是一款开源的分布式图数据库，自 2019 年开源以来，先后被美团、京东、360 数科、快手、众安金融等多家企业采用，应用在智能推荐、金融风控、数据治理、知识图谱等等应用场景。GitHub 地址：https://github.com/vesoft-inc/nebula

无论最终性能测试在什么环境、待测图数据库是什么，只要 Driver 的输入（数据和配置）相同，其生成的 Workload 就是相同的。为了得到最高的吞吐量，一般会把 Workload 拆成多个 Stream。对于图来说，多个操作是有前后依赖关系的，比如好友存在才能添加好友关系。

LDBC SNB 系列：深入理解 Benchmark 的 Query 生成和执行

![nebula-br local-store](https://nebula-website-cn.oss-cn-hangzhou.aliyuncs.com/nebula-website-5.0/share/2024-04-02-1.webp.jpg)

看过我之前文章的读者可能知道我之前用 nebula-br 搭建过主备集群，具体实践过程不做赘述，有兴趣的小伙伴可以看下这个链接：nebula-br local-store 模式搭建主备集群实践 🔗

那么问题来了，为什么在已有 nebula-br 搭建集群的情况下，我又造了个“新轮子”，让 spark-connector 走上 nebula-br 的老路呢？在这里我来讲讲二者的优劣势：

**！注意**
文章内容仅代表个人观点，仅供参考。

## 主备集群搭建的选择

先从目前个人已知的可搭建备集群的开箱即用工具讲起，NebulaGraph 社区的相关搭建工具有：

- nebula-br
- nebula-spark-connector
- nebula-flink-connector

由于 nebula-flink-connector 和 nebula-spark-connector 实践原理较为相似，本文就不单独地讲述 nebula-flink-connector 如何搭建主备集群。

### 搭建备集群实践

nebula-br 主备集群是通过本地生成 snapshot 备份，然后传输到备集群进行恢复的方式。

- 优点：
  - 搭建速度快；

- 缺点：
  - 生成备份文件的时候会导致 Nebula 不可写，影响在线业务；
  - 环境搭建麻烦，需要在主备集群上运行 Agent 和 BR，此外恢复时需要读取备份文件，这些文件要么通过 S3 或者 NTFS 共享，要么就手动拷贝；
  - 从源集群备份还原到目标集群，目标集群将会被覆盖（也就是跟源集群保持一致，可以理解为目标集群将会被清空，然后还原源集群数据到目标集群）。

### spark-connector 搭建集群的优劣势

nebula-spark-connector 主备集群是通过 nebula-client 去 scan nebula 的数据，然后生成 INSERT 语句在备集群进行重放。

- 优点：
  - 开箱即用，针对小数据集群（低于 1 亿）基本上不需要任何开发直接使用；大数据集群（大于 1 亿）可能需要额外流量控制逻辑；
  - 同步时对源集群的读写影响较小（前提是需要控制导出并发控制）；
  - 同步粒度更精细，可以指定同步 Space、Tag、Edge；

- 缺点：
  - 导出导入的方式同步数据比较慢。就速度这点和官方交流了下，官方表示之前在某家公司业务上实测过速度，nebula-spark-connector 做同步是要比 nebula-br 快 3-5 倍。可能环境和参数有差异，如果读者你选择工具的时候，可以先测试下相关工具的性能数据。
  - 需要 Spark 环境支持；
  - 无法中断重启时接着上次 cursor 继续 scan；

这里稍微说下 nebula-flink-connector，个人感觉它跟 nebula-spark-connector 最大的区别在于 Spark 需要读完整个分区才会做 Write 操作，nebula-flink-connector 可以做到边读边写，理论上效率会高一点。

**nebula-spark-connector 原理**

![nebula-spark-connector](https://nebula-website-cn.oss-cn-hangzhou.aliyuncs.com/nebula-website-5.0/share/2024-04-02-2.webp.jpg)

本图有任何理解错误的地方，麻烦在留言帮忙指出，感谢。


## 搭建备集群实践

nebula-spark-connector 的 GitHub 仓库地址：https://github.com/vesoft-inc/nebula-spark-connector，本文采用 nebula-spark-connector 的 release-3.6 分支，并以 Spark 3.x 为例。

首先，准备相关的运行环境：
```
`$ git clone -b release-3.6 https://github.com/vesoft-inc/nebula-spark-connector.git
$ cd nebula-spark-connector
$ mvn clean package -Dmaven.test.skip=true -Dgpg.skip -Dmaven.javadoc.skip=true -pl nebula-spark-connector_3.0,example -am -Pscala-2.12 -Pspark-3.0`
```

NebulaGraph 的导入导出可以直接使用 example 中的 Nebula2Nebula.scala 完成，通过上面 maven 打包命令后在 example 的 target 目录可得到 example-3.0-SNAPSHOT-jar-with-dependencies.jar 文件，提交 Spark 任务命令：

```
`spark-submit --master local \
 --conf spark.driver.extraClassPath=./ \
 --conf spark.executor.extraClassPath=./  \
 --class com.vesoft.nebula.examples.connector.Nebula2Nebula \
 example-3.0-SNAPSHOT-jar-with-dependencies.jar \
 -sourceMeta "xxxxx:9559" -targetMeta "yyyyy:9559" -targetGraph "yyyy:9669" -sourceSpace "source" -targetSpace "target" -includeTag "user" -excludeTags "aa,bb" -excludeEdges "cc,dd"  -limit 2  -batch 2 -p 8 -timeout 50000 -u root -passwd nebula`
```

参数解释：

- sourceMeta：源 nebula 服务 metad 地址（从哪里导出）
- targetMeta：目标 nebula 服务 metad 地址
- targetGraph：目标 nebula 服务 graphd 地址，多个地址以英文逗号隔开（导出数据重新组装 INSERT 语句将通过此地址进行提交）
- sourceSpace：源 nebula 的 Space，也就是要导出的 Space 名称
- targetSpace：目标 nebula 的 Space，也就是要导入的 Space 名称
- includeTag：指定导出的 Tag，不指定就全部 Tag 导出
- excludeTags：排除哪些 Tag 不进行导出同步
- excludeEdges：排除哪些 Edge 不进行导出同步
- limit：从源 nebula 导出时每次 scan 多少数据，值越大单次 scan 的数据量越大，对源 nebula 服务器的压力越大，在没有达到磁盘和网络 IO 的瓶颈前，理论上值越大导出更快
- batch：导出的数据多少一批打包成一条 INSERT 语句发送到目标 nebula 服务的 graphd 机器上进行执行
- p：也叫 writeParallel, 导出的数据拆分成多少个任务并行写，值越大速度越快，但是目标 nebula 服务器压力越大
- timeout：每次 scan 源数据的超时时间

### 优化改造

这里讲下为什么要做这些优化改造：

1.经我们线上验证，Nebula2Nebula 示例是一次所有 Partition 导出后才做导入操作，对于数据量非常大的情况下并不友好、效率太低，此外对 Spark 来说会占用太多的内存可能会导致内存不足；不过，官方表示他们不是所有的 Partition 导出之后才进行导入，而是边读边写，且消耗内存很小。因为我这边已经完成了相关的开发工作，边读边写的操作就留给读者你来实践了。

2.无法支持源 NebulaGraph storaged 地址为 127.0.0.1 的场景（一般属于单机场景），因为 Spark 任务导出的 storaged 地址是通过 metad 元数据获取，就会导致导出任务会请求 127.0.0.1 的 storaged；

3.缺少导出和导入统计，不知道当前已经导出多少数据量；
下面开始进入改造（解决上述问题）部分：

### 导出优化

本次的优化改造的导出优化：将 Partition 进行拆分，改成多线程的方式，每个线程处理一个 Partition。

Nebula2Nebula.scala 改造：增加 readParallel 参数，用于指定多少个线程可同时读取 nebula，控制读取速率，不然可能会对线上服务器照成影响：

```
`val readParallelOption = new Option("rp", "readParallelNum", true, "parallel for read data")
options.addOption(readParallelOption)

val readParallelNum: Int =
      if (cli.hasOption("readParallelNum")) cli.getOptionValue("readParallelNum").toInt else partitions
for (partitionId <- 1 to readPartition) {
      val task = new Runnable {
        def run(): Unit = {
          syncTagPartitionData(spark,
            sourceConfig,
            sourceSpace,
            limit,
            readPartition,
            targetConfig,
            targetSpace,
            batch,
            tag,
            writeParallel,
            user,
            passwd,
            overwrite,
            partitionBatch,
            eachScanWaitTime,
            partitionId
          )
        }
      }
      threadPool.execute(task);
    }`
```

NebulaOptions.scala 改造：增加参数可指定 Partition 读取，代码示例：

```
`var readPartitionId: Int = parameters.getOrElse(READ_PARTITION_ID, 0).toString.toInt`
```


SimpleScanBuilder.scala 改造：


```
`override def planInputPartitions(): Array[InputPartition] = {
 //读取指定的分区
  Array(NebulaPartitionBatch(Array(nebulaOptions.readPartitionId)))
}
case class NebulaPartitionBatch(partitions: Array[Int]) extends InputPartition`
```


NebulaPartitionReaderFactory.scala 改造：读取指定分区数据，代码示例：

```
`class NebulaPartitionReaderFactory(private val nebulaOptions: NebulaOptions,
                                   private val schema: StructType)
    extends PartitionReaderFactory {

  private val LOG: Logger = LoggerFactory.getLogger(this.getClass)

  override def createReader(inputPartition: InputPartition): PartitionReader[InternalRow] = {
    val partitions = inputPartition.asInstanceOf[NebulaPartitionBatch].partitions
    if (DataTypeEnum.VERTEX.toString.equals(nebulaOptions.dataType)) {
      LOG.info(s"Create NebulaVertexPartitionReader partitions:${partitions}")
      new NebulaVertexPartitionReader(partitions, nebulaOptions, schema)
    } else {
      LOG.info(s"Create NebulaEdgePartitionReader partitions:${partitions}")
      new NebulaEdgePartitionReader(partitions, nebulaOptions, schema)
    }
  }
}`

```

### 支持本地地址

下面开始支持导出 storaged 的地址为 127.0.0.1，示例代码参见：

Nebula2Nebula 改造：支持配置源 nebula 的 storaged 连接地址，示例代码见下：

```
`val sourceStorageOption =
      new Option("sourceStorage", "sourceStorageAddress", true, "source nebulagraph storage address")
options.addOption(sourceStorageOption)
NebulaReader.scala 改造：替换原来的 StorageClient，示例代码见下：
this.storageClient = new SpecialStorageClient(nebulaOptions, address.asJava, nebulaOptions.timeout)`
```

增加 SpecialStorageClient.scala: 用于获取 storaged 地址的时候返回指定的 storageAddress，示例代码见下：

```
`class SpecialStorageClient(nebulaOptions: NebulaOptions, addresses: util.List[HostAddress], timeout: Int = 10000, connectionRetry: Int = 3,
                           executionRetry: Int = 3, enableSSL: Boolean = false, sslParam: SSLParam = null)
  extends StorageClient(addresses, timeout, connectionRetry, executionRetry, enableSSL, sslParam) {

  private val LOG: Logger = LoggerFactory.getLogger(this.getClass)

  override def connect(): Boolean = {
    val success = super.connect()
    if (!success) {
      return success
    }
    val storageAddress = nebulaOptions.getStorageAddress
    if (storageAddress.isEmpty) {
      return success
    }
    setStorageAddress(storageAddress)
    LOG.info(s"Set special storage address:${storageAddress.asJava} | value:${nebulaOptions.storageAddress}")
    success
  }

  private def setStorageAddress(storageAddress: ListBuffer[HostAddr]) = {
    val clazz = this.getClass
    var field = classOf[StorageClient].getDeclaredField("metaManager")
    field.setAccessible(true)
    var metaManager = new SpecialStorageMetaManager(addresses, timeout, connectionRetry, executionRetry, enableSSL, sslParam)
    metaManager.setStorageAddress(storageAddress)
    field.set(this, metaManager)
  }
}

class SpecialStorageMetaManager(address: util.List[HostAddress], timeout: Int, connectionRetry: Int,
                                executionRetry: Int, enableSSL: Boolean, sslParam: SSLParam) extends
  MetaManager(address, timeout, connectionRetry, executionRetry, enableSSL, sslParam) {

  private val LOG: Logger = LoggerFactory.getLogger(this.getClass)

  var storageAddress: ListBuffer[HostAddr] = null;

  override def getPartsAlloc(spaceName: String): util.Map[Integer, util.List[HostAddr]] = {
    val partMap: util.Map[Integer, util.List[HostAddr]] = getPartsAlloc(spaceName)
    if (storageAddress != null && !storageAddress.isEmpty) {
      partMap.keySet().forEach(partId => {
        LOG.error(s"Storage partId:${partId} address replace fix value:${storageAddress.asJava}")
        partMap.put(partId, storageAddress.asJava)
      })
    }
    partMap
  }

  override def listHosts(): util.Set[HostAddr] = {
    if (storageAddress != null && !storageAddress.isEmpty) {
      LOG.error(s"[listHosts] Storage address replace fix value:${storageAddress.asJava}")
      return storageAddress.toSet.asJava
    }
    super.listHosts()
  }


  override def getLeader(spaceName: String, part: Int): HostAddr = {
    if (storageAddress != null && !storageAddress.isEmpty) {
      LOG.error(s"[getLeader] Storage address replace fix value:${storageAddress.asJava}")
      return storageAddress.last
    }
    super.getLeader(spaceName, part)
  }

  def setStorageAddress(storageAddress: ListBuffer[HostAddr]): Unit = {
    this.storageAddress = storageAddress
  }
}`
```


### 支持导入打印统计

新增导出导入统计打印，方便知道当前进度。

增加 ReadWriteStats.scala：用于统计，示例代码见下：

```
`object ReadWriteStats {

  private var readCounter = new ConcurrentHashMap[String, Int]();
  private var writeCounter = new ConcurrentHashMap[String, Int]();

  def incrementReadCount(key: String, num: Int): Int = {
    readCounter.compute(key, (_, count) => count + num)
  }

  def getReadCount(key: String): Int = {
    readCounter.getOrDefault(key, 0)
  }

  def incrementWriteCount(key: String, num: Int): Int = {
    writeCounter.compute(key, (_, count) => count + num)
  }

  def getWriteCount(key: String): Int = {
    writeCounter.getOrDefault(key, 0)
  }

}`
```


NebulaReader.scala 改造：增加 scan 读取统计数据输出，示例代码见下：

```
`protected def getRow(): InternalRow = {
    ......
    val totalCount = ReadWriteStats.incrementReadCount(nebulaOptions.label, 1)
    if (totalCount > 10000 && totalCount % 10000 == 0) {
      LOG.info(s"Reader vertex:${nebulaOptions.label} , totalCount:${totalCount}")
    }
    mutableRow
  }

protected def hasNextVertexRow: Boolean = {
    .....
    if (!hashNext) {
      LOG.info(s"Finish read vertex:${nebulaOptions.label} , readTotalCount:${ReadWriteStats.getReadCount(nebulaOptions.label)}")
    }
    hashNext
  }  

protected def hasNextEdgeRow: Boolean = {
    ....
    if (!hashNext) {
      LOG.info(s"Finish read edge:${nebulaOptions.label} , readTotalCount:${ReadWriteStats.getReadCount(nebulaOptions.label)}")
    }
    hashNext
  }`
```

NebulaVertexWriter.scala 改造：增加写入统计数据输出，示例代码见下：

```
`def execute(): Unit = {
    val nebulaVertices = NebulaVertices(propNames, vertices.toList, policy)
    val exec = nebulaOptions.writeMode match {
      case WriteMode.INSERT =>
        val vertexCount = nebulaVertices.values.length
        val totalCount = ReadWriteStats.incrementWriteCount(nebulaOptions.label,vertexCount)
        if (totalCount > nebulaOptions.printRWRowCount && totalCount % nebulaOptions.printRWRowCount == 0) {
          LOG.info(s"Batch insert vertex:${nebulaOptions.label} , count: ${vertexCount} , totalCount:${totalCount}")
        }
        ...
    }
}  

override def commit(): WriterCommitMessage = {
    if (vertices.nonEmpty) {
      execute()
    }
    LOG.error(s"Write vertex:${nebulaOptions.label} task finished. Haven write totalCount:${ReadWriteStats.getWriteCount(nebulaOptions.label)}")
    .....
}`
```


改造完毕之后重新使用 maven 命令打包，再使用 Spark 命令提交任务即可。

这里打包了相关的代码：请点击阅读原文查看，如果你有需要可以拷贝到你的运行环境，自行替换。

以上，感谢你的阅读。

如果本文有任何技术错误，麻烦评论指出，谢谢你的反馈。

## 关于 NebulaGraph  

NebulaGraph 是一款开源的分布式图数据库，自 2019 年开源以来，先后被美团、京东、360 数科、快手、众安金融等多家企业采用，应用在智能推荐、金融风控、数据治理、知识图谱等等应用场景。GitHub 地址：https://github.com/vesoft-inc/nebula

nebula-br 主备集群是通过本地生成 snapshot 备份，然后传输到备集群进行恢复的方式。nebula-spark-connector 主备集群是通过 nebula-client 去 scan nebula 的数据，然后生成 INSERT 语句在备集群进行重放。