Spark Structured Streaming写入Kafka压缩消息失败配置咨询

Spark Structured Streaming 3.2.1支持向Kafka写入压缩消息，你遇到的“compression.type属性未知”报错，核心原因是Spark依赖的Kafka客户端版本与你的Kafka 0.10集群版本兼容性问题，以及配置细节需要调整。

问题分析 #

Spark 3.2.1默认绑定的kafka-clients版本是2.8.0，与Kafka 0.10集群的版本差距过大，高版本客户端的部分属性传递到低版本broker时会出现识别异常。另外需明确：Kafka从0.9版本开始就支持snappy、gzip、lz4等压缩算法，Kafka 0.10本身是满足snappy压缩支持的。

正确配置方式 #

1. 调整Kafka客户端依赖版本
   在你的Spark项目中，将kafka-clients的版本降级到与Kafka 0.10匹配的版本（例如0.10.2.2），避免版本兼容性问题。以Maven为例，在pom.xml中添加依赖排除与指定：

   <dependency>
       <groupId>org.apache.spark</groupId>
       <artifactId>spark-sql-kafka-0-10_2.12</artifactId>
       <version>3.2.1</version>
       <exclusions>
           <exclusion>
               <groupId>org.apache.kafka</groupId>
               <artifactId>kafka-clients</artifactId>
           </exclusion>
       </exclusions>
   </dependency>
   <dependency>
       <groupId>org.apache.kafka</groupId>
       <artifactId>kafka-clients</artifactId>
       <version>0.10.2.2</version>
   </dependency>
   

2. 正确配置压缩参数
   保持kafka.compression.type的配置写法，但需确保DataFrame包含Kafka连接器要求的key和value列（若不需要key，可设为null）。修正后的代码示例：

   import org.apache.spark.sql.streaming.Trigger
   
   // 确保finalDf包含key和value列，示例：
   // val finalDf = rawDf.selectExpr("CAST(id AS STRING) AS key", "CAST(data AS STRING) AS value")
   
   var dataStreamWriter = 
       finalDf
      .writeStream
      .format("kafka")
      .option("topic", topic)
      .option("kafka.compression.type", "snappy") // 正确的压缩配置
      .option("kafka.batch.size", "1024000") // 批量大小配合压缩提升效率
      .option("checkpointLocation", checkpointLocation)
      .trigger(Trigger.ProcessingTime(s"${triggerDuration} seconds"))
      .start() // 必须调用start()启动流任务
   

3. 验证压缩效果
   可以通过Kafka命令行工具验证消息是否被压缩：

   kafka-console-consumer.sh --bootstrap-server <你的broker地址> --topic <目标主题> --property print.key=true --property print.value=false --property print.metadata=true
   

   查看输出中的compression.type字段，确认是否为snappy。

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