如何限制Spark（Databricks）写入Neo4j时的线程数量？

不用缩小集群规格，你可以通过以下几种方式控制写入Neo4j的线程数量：

• 调整Neo4j Spark Connector的并发参数
  这个连接器本身提供了控制每个executor写入并发的参数neo4j.batch.concurrent.writes，默认值可能较高（比如对应你看到的200线程），直接在写入配置里添加这个参数就能降低单executor的并发线程数：

  edges.write.format("org.neo4j.spark.DataSource")\
  .option("url", "neo4j://url:7687") \
  .mode("overwrite")\
  .option("relationship", "connected")\
  .option("batch.size",1000)\
  .option("neo4j.batch.concurrent.writes", 10)  # 按需设置每个executor的并发线程数
  .option("relationship.save.strategy", "keys")\
  .option("relationship.source.node.keys", "id:id")\
  .option("relationship.target.node.keys", "id:id")\
  .option("relationship.source.labels", "node")\
  .option("relationship.target.labels", "node")\
  .save()
  

  这个参数直接限制连接器在每个executor内启动的写入线程数，是最直接的控制方式。

• 减少Spark任务并行度
  你之前设置的spark.executor.cores没生效，可能是因为DataFrame的分区数太多，导致同时运行的任务数远超executor核心数。可以通过重分区减少任务数量，间接降低总并发线程数：

  # 先把DataFrame分区数调整到合适值，比如20
  edges_repartitioned = edges.repartition(20)
  edges_repartitioned.write.format("org.neo4j.spark.DataSource")\
  # 后续配置同之前代码
  .save()
  

  也可以通过设置spark.sql.shuffle.partitions（如果写入前有shuffle操作）来控制分区数，默认是200，这可能就是你看到200线程的原因之一。

• 配合Spark资源配置
  单独设置spark.executor.cores不够，要结合其他资源参数一起调整：

  • 设置spark.executor.cores=4，同时保持spark.task.cpus=1（默认值），这样每个executor最多同时运行4个任务
  • 设置spark.max.executor.instances限制executor的总数量，避免过多executor同时连接Neo4j
    这些配置可以在SparkSession初始化时设置：

  from pyspark.sql import SparkSession
  
  spark = SparkSession.builder \
      .appName("Neo4jWrite") \
      .config("spark.executor.cores", "4") \
      .config("spark.max.executor.instances", "5") \
      .getOrCreate()
  

• Neo4j端辅助限制（可选）
  如果你需要双重保险，可以在Neo4j的neo4j.conf里设置dbms.connections.max.outbound来限制全局出站连接数，但这个会影响所有连接Neo4j的应用，优先从Spark端调整更合理。

你之前设置spark.executor.cores无效的核心原因是：Neo4j Spark Connector在每个Spark任务内部可能会启动多个线程，光限制executor核心数只能控制同时运行的任务数，没法管控每个任务内连接器启动的线程，必须结合连接器自身的并发参数才能精准控制总线程数。

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