Pick of the Week'22 ｜第 4 周看点社区数据公布

文章 NebulaGraph
1

每周看点

每周五 Nebula 为你播报每周看点，每周看点由固定模块：产品动态、社区问答、推荐阅读，和随机模块：本周大事件构成。

天气渐冷，年味渐浓，不知道本周搬砖的你状态如何，来查收下本周的 Nebula 动态吧~~

本周看点

Nebula 社区数据公布

本周 Nebula 公布了社区数据报告以及社区年度 nStar 盘点。

产品动态

本周 Nebula 主要有这些产品动态：

社区问答

Pick of the Week 每周会从官方论坛、知乎、微信群、微信公众号及开源中国等渠道精选问题同你分享。

主题分享

本周分享的主题是【如何快速导入 Dgraph 数据】，由社区用户 bjzhaoqing 提出，Nebula 研发解答。

bjzhaoqing 提问：有没有工具可以快速导入 Dgraph 数据到 Nebula 进行测试？我看 Nebula 可以导入 csv 格式数据。我用什么方式可以将 g01.rdf 转换成 csv 格式呢？”

Nebula：有两种方式：

以导入 MySQL 数据源为例（只需要自己写读取数据源的几行代码，写入 Nebula 的代码同理）

def writeVertex(spark: SparkSession): Unit = {
    LOG.info("start to write nebula vertices")
    var df = spark.read
      .format("jdbc")
      .option("url", "jdbc:mysql://127.0.0.1:3360/db?useUnicode=true&characterEncoding=utf-8")
      .option("dbtable", "table")
      .option("user", "root")
      .option("password", "nebula")
      .load()
      
    df = df.select("select id, a as name, b as age from table")

    val config = getNebulaConnectionConfig()
    val nebulaWriteVertexConfig: WriteNebulaVertexConfig = WriteNebulaVertexConfig
      .builder()
      .withSpace("test")
      .withTag("person")
      .withVidField("id")
      .withVidAsProp(false)
      .withBatch(1000)
      .build()
    df.write.nebula(config, nebulaWriteVertexConfig).writeVertices()
}

Nebula 进阶技能

本周的 Nebula 进阶技能分享一个【nebula-python 多线程慢的小实验】，来源于 Harris Chu 的分享：https://discuss.nebula-graph.com.cn/t/topic/7285。

论坛里经常有人问起 nebula-python 多线程慢的问题，用一个小实验说明一下。

nebula-python 主要是 2 个问题：

  1. nebula-python 解码是使用 Python 原生代码来解码的，解码效率会比较慢。
  2. 受限于 Python 的全局解释锁，nebula-python 多线程时最多只能用到1 个 CPU。

两个问题叠加起来的现象就是，如果查询有比较多的数据，那多线程效率是不会提升的，因为单核的 CPU 已经被打满。甚至有可能因为多线程的竞争，比不用多线程更慢一些。

下面是实验的脚本，使用 LDBC sf30 的数据，分别用普通代码，多线程，多进程运行 10 次语句。

# case 1, 语句
go 2 steps from 2608 over KNOWS yield KNOWS._dst, KNOWS.creationDate as creationDate

返回条数 118,567，服务端 latency 在 232,645 us 左右。单次执行耗时在 3 秒左右。

normalExecutor: running in 29.527837991714478 seconds
threadExecutor: running in 29.018051624298096 seconds
processExecutor: running in 15.147454977035522 seconds

# case 2 语句
go 2 steps from 2608 over KNOWS yield KNOWS._dst, KNOWS.creationDate as creationDate | limit 1000

返回条数 1,000，服务端 latency 在 103,957 us 左右。

normalExecutor: running in 1.8793082237243652 seconds
threadExecutor: running in 0.28518009185791016 seconds
processExecutor: running in 0.27771878242492676 seconds

代码如下：

import time
import concurrent
from nebula2.Config import Config
from nebula2.gclient.net import ConnectionPool


worker_num = 10
sessions = []
connection_pool = ConnectionPool()


def run(stmt, index):
    session = sessions[index]
    resp = session.execute(stmt)
    return resp


def threadExecutor(stmt):
    with concurrent.futures.ThreadPoolExecutor(max_workers=worker_num) as executor:
        runner = []
        start = time.time()
        for i in range(worker_num):
            future = executor.submit(run, stmt, i)
            runner.append(future)
        for future in concurrent.futures.as_completed(runner):
            if future.exception() is not None:
                raise future.exception()
            else:
                rs = future.result()
                if rs is not None:
                    print("row size is {}".format(rs.row_size()))
                    print("latency is {}".format(rs.latency()))
    print("threadExecutor: running in {} seconds".format(time.time() - start))


def processExecutor(stmt):
    start = time.time()

    with concurrent.futures.ProcessPoolExecutor(max_workers=worker_num) as executor:
        runner = []
        start = time.time()
        for i in range(worker_num):
            future = executor.submit(run, stmt, i)
            runner.append(future)
        for future in concurrent.futures.as_completed(runner):
            if future.exception() is not None:
                raise future.exception()
            else:
                rs = future.result()
                if rs is not None:
                    print("row size is {}".format(rs.row_size()))
                    print("latency is {}".format(rs.latency()))
    print("processExecutor: running in {} seconds".format(time.time() - start))


def normalExecutor(stmt):
    start = time.time()
    for i in range(worker_num):
        rs = run(stmt, i)
        print("row size is {}".format(rs.row_size()))
        print("latency is {}".format(rs.latency()))
    print("normalExecutor: running in {} seconds".format(time.time() - start))


if __name__ == "__main__":
    stmt = "go 2 steps from 2608 over KNOWS yield KNOWS._dst, KNOWS.creationDate as creationDate| yield count(*)"
    config = Config()
    config.max_connection_pool_size = worker_num
    assert connection_pool.init([("192.168.15.13", 1669)], config)
    for _ in range(worker_num):
        sess = connection_pool.get_session("root", "nebula")
        sess.execute("use sf30")
        sessions.append(sess)

    normalExecutor(stmt)
    threadExecutor(stmt)
    processExecutor(stmt)

结论：

  • 当查询的数据比较多的时候，受限于全局解释性锁，多线程并不会比串行执行好很多，因为解码这个时候是 CPU 密集；
  • 当查询的数据比较少的时候，尤其是 count(*) 比较明显，因为这个时候是 IO 等待了，多线程会比非多线程快很多。

推荐阅读

  • 《详解 Nebula Dashboard 企业版多集群监控运维工具》
    • 推荐理由：数据库的运维工作至关重要，及时发现及时应对，而 Nebula Dashboard 提供可视化的监控功能让你实时了解集群情况。公众号回复「试用」占限量试用名额抢先体验企业版 Nebula Dashboard~

星云·小剧场

为什么给图数据库取名 Nebula？

Nebula 是星云的意思，很大嘛，也是漫威宇宙里面漂亮的星云小姐姐。对了，Nebula 的发音是：[ˈnɛbjələ]

本文星云图讲解–《猎户星座 M78》

猎户星座 M78

在富饶的猎户座里，M78 是其中最明亮者之一，泛蓝色反射星云 M78 离我们约 1,500 光年远，大小则在 5 光年左右。它的色泽，来自它内部的尘埃偏好反射附近炽热年轻亮星之蓝色星光。紧贴在 M78 左侧的反射星云，为 NGC 2071。

影像提供与版权：Wes Higgins
作者与编辑：Robert Nemiroff (MTU) & Jerry Bonnell (UMCP)

这是一个从 https://nebula-graph.com.cn/posts/nebula-graph-weekly-pickup-2022-01-21/ 下的原始话题分离的讨论话题