Код Spark для записи в тему, сжатую Кафкой

Question

Кто-нибудь знает, можно ли назвать writeStream темой Кафки, где создаваемая тема является уплотненной? Приведенный ниже код создает тему Kafka, но переданные параметры игнорируются.

    StreamingQuery query = ds
            .writeStream()
            .format("kafka")
            .option("kafka.bootstrap.servers", "localhost:9092"))
            .option("topic", "myTopic")
            .option("cleanup.policy", "compact")
            .option("min.insync.replicas", 1)
            .option("segment.bytes", 4096)
            .option("delete.retention.ms", 100)
            .option("min.compaction.lag.ms", 0)
            .option("min.cleanable.dirty.ratio", 0.01)
            .start();

Answer 1

Spark Structure Streaming, когда звонки writeStream с форматом кафки под капотом используют KafkaProducer.

Если на стороне брокера auto.create.topics.enable имеет true в качестве значения (что должно быть по умолчанию), Когда KafkaProducer выбирает метаданные, брокер создает новую тему.

KafkaProducer не передает свойств темы . Брокер использует значение по умолчанию один раз (например, default.replication.factor, num.partitions, log.cleanup.policy) для создания темы.

Вы не можете передавать свойства создания темы через KafkaProducer.

Answer 2

Как описано в документации, также используйте префикс kafka со свойствами производителя или потребителя.

Собственные конфигурации Кафки можно установить через DataStreamReader.option с kafka.префикс.

Пример:

stream.option("kafka.bootstrap.servers","host:port")

Кроме того, в .options можно передавать только свойства уровня производителя или потребителя.Если вам нужно установить какую-либо конфигурацию на уровне темы, вам нужно использовать конфигурацию посредника, которую нельзя настроить с помощью кода потоковой передачи, который выбирается из конфигурации по умолчанию.

https://spark.apache.org/docs/2.2.0/structured-streaming-kafka-integration.html#kafka-specific-configurations