如何减慢Kafka Producer的写入速度？答案

【问题标题】：How to slow down the write speed of Kafka Producer?如何减慢Kafka Producer的写入速度？
【发布时间】：2019-11-04 20:23:03
【问题描述】：

我用spark就是这样写数据到kafka的。

df.write()。格式（“卡夫卡”）。保存（）

我可以控制写入kafka的速度以避免对kafka造成压力吗？是否有一些选项有助于降低速度？

【问题讨论】：

标签： apache-spark apache-kafka spark-structured-streaming

【解决方案1】：

我认为将linger.ms 设置为非零值会有所帮助。因为它控制在发送当前批次之前等待附加消息的时间量。代码如下所示

df.write.format("kafka").option("linger.ms", "100").save()

但这真的取决于很多事情。如果您的 Kafka 足够“大”并且配置正确，那么我不会太担心速度。毕竟，kafka 就是为了应对这种情况（流量高峰）而设计的。

【讨论】：

此选项是否适用于 spark 结构化流？
对不起，我是从Kafka的角度回答这个问题，因为我比较熟悉。您必须尝试找出它是否有效。

【解决方案2】：

一般情况下，结构化流式处理会在默认情况下尝试尽可能快地处理数据。每个源中都有允许控制处理速率的选项，例如 File 源中的 maxFilesPerTrigger 和 Kafka 源中的 maxOffsetsPerTrigger。

val streamingETLQuery = cloudtrailEvents
  .withColumn("date", $"timestamp".cast("date") // derive the date
  .writeStream
  .trigger(ProcessingTime("10 seconds")) // check for files every 10s
  .format("parquet") // write as Parquet partitioned by date
  .partitionBy("date")
  .option("path", "/cloudtrail")
  .option("checkpointLocation", "/cloudtrail.checkpoint/")
  .start()

val df = spark.readStream
  .format("text")
  .option("maxFilesPerTrigger", 1)
  .load("text-logs")

【讨论】：

感谢您的回答！但我必须编辑我的问题，它应该是火花而不是结构化流。你的答案似乎试图控制进程速率，而我实际上试图减慢 Spark 中的kafka producer。