NebulaGraph数据库：如何使用PySpark的Spark Connector写入数据？

当然存在用Python（PySpark）向NebulaGraph写入数据的方案，依托nebula-spark-connector就能实现，和Scala版本的逻辑一致，Python环境下完全兼容。

前提准备 #

启动PySpark时需要指定connector的jar包，命令如下：

/spark/bin/pyspark --driver-class-path nebula-spark-connector-3.0.0.jar --jars nebula-spark-connector-3.0.0.jar

读取数据示例（你提供的代码） #

先确认读取数据的逻辑是可行的，代码如下：

df = spark.read.format(
  "com.vesoft.nebula.connector.NebulaDataSource").option(
    "type", "vertex").option(
    "spaceName", "basketballplayer").option(
    "label", "player").option(
    "returnCols", "name,age").option(
    "metaAddress", "metad0:9559").option(
    "partitionNumber", 1).load()

写入顶点数据示例 #

假设你已经有了要写入的顶点DataFrame（比如包含player_id、name、age字段），可以按以下方式写入：

# 假设df_vertex是准备好的顶点数据DataFrame
df_vertex.write.format("com.vesoft.nebula.connector.NebulaDataSource") \
    .option("type", "vertex") \
    .option("spaceName", "basketballplayer") \
    .option("label", "player") \
    .option("metaAddress", "metad0:9559") \
    .option("graphAddress", "graphd0:9669") \
    .option("vertexId", "player_id") \  # 指定DataFrame中对应顶点ID的字段
    .option("properties", "name,age") \  # 指定要写入的属性字段
    .option("batch", 1000) \  # 批量写入的大小
    .mode("append") \  # 写入模式：append/overwrite等
    .save()

写入边数据示例 #

如果要写入边数据（比如follow类型的边，包含src_player、dst_player、degree字段），代码如下：

# 假设df_edge是准备好的边数据DataFrame
df_edge.write.format("com.vesoft.nebula.connector.NebulaDataSource") \
    .option("type", "edge") \
    .option("spaceName", "basketballplayer") \
    .option("label", "follow") \
    .option("metaAddress", "metad0:9559") \
    .option("graphAddress", "graphd0:9669") \
    .option("srcId", "src_player") \  # 边的起始顶点ID字段
    .option("dstId", "dst_player") \  # 边的目标顶点ID字段
    .option("properties", "degree") \  # 边的属性字段
    .option("batch", 1000) \
    .mode("append") \
    .save()

注意事项 #

• 确保nebula-spark-connector的版本与你的NebulaGraph集群版本匹配，避免兼容性问题
• graphAddress需要指定NebulaGraph的graphd服务地址，写入操作必须依赖该参数
• 批量大小batch可以根据数据量调整，提升写入效率
• 写入模式mode根据需求选择：append是追加数据，overwrite会先清空对应标签的数据再写入

内容的提问来源于stack exchange，提问作者user19733404