Как замедлить скорость записи Kafka Producer?

Question

Я использую искру для записи данных в kafka таким способом.

df.write (). Формат ( "Кафка"). save ()

можно ли контролировать скорость записи в кафку, чтобы избежать давления на кафку? Есть ли варианты, которые помогают снизить скорость?

Answer 1

Как правило, структурированная потоковая передача попытается обработать данные максимально быстро по умолчанию. В каждом источнике есть опции, позволяющие контролировать скорость обработки, такие как maxFilesPerTrigger в источнике файла и 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")

Для получения более подробной информации прочитайте следующие ссылки:

https://jaceklaskowski.gitbooks.io/spark-structured-streaming/spark-sql-streaming-KafkaSource.html https://jaceklaskowski.gitbooks.io/spark-structured-streaming/spark-sql-streaming-FileStreamSource.html https://databricks.com/blog/2017/01/19/real-time-streaming-etl-structured-streaming-apache-spark-2-1.html http://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#input-sources http://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html

Answer 2

Я думаю, что установка linger.ms в ненулевое значение поможет. Так как он контролирует количество времени ожидания дополнительных сообщений перед отправкой текущего пакета. Код может выглядеть следующим образом

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

Но это действительно зависит от многих вещей. Если ваш Kafka достаточно «большой» и настроен правильно, я бы не стал сильно беспокоиться о скорости. В конце концов, kafka разработан, чтобы справиться с этой ситуацией (всплеск трафика).