我正在旋转一个emr来向kafka发布一个Dataframe（大约300-400行）。我可以发布它，Dataframe有200个分区。在发布Dataframe时，我看到kafka集群中的cpu出现了大约20-30分钟的巨大峰值。分区号是否创建了200个连接？
或者像这里所说的那样使用1个连接。http://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html#producer-缓存
样本代码

spark-shell --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.0

import org.apache.spark.sql.functions.col
val kafkaOptions = Map("kafka.bootstrap.servers" -> s"$host:$port",
        "kafka.security.protocol" -> "SSL",
        "kafka.ssl.endpoint.identification.algorithm" -> "",
        "kafka.ssl.truststore.location" -> "/home/hadoop/client.truststore.jks",
        "kafka.ssl.truststore.password" -> "password",
        "kafka.ssl.keystore.type" -> "PKCS12",
        "kafka.ssl.key.password" -> "password",
        "kafka.ssl.keystore.location" -> "/home/hadoop/client.keystore.p12",
        "kafka.ssl.keystore.password" -> "password")
    )

 val df = spark
        .read
        .option("header", true)
        .option("escape", "\"")
        .csv("s3://bucket/file.csv")

 val publishToKafkaDf = df.withColumn("value", col("body"))

 publishToKafkaDf
      .selectExpr( "CAST(value AS STRING)")
      .write
      .format("kafka")
      .option("topic", "test-topic")
      .options(kafkaOptions)
      .save()