在ConFluent S3 Kafka连接器中压缩Avro数据

Question

我有一个 Confluent 接收器连接器，它正在从 Kafka 主题获取数据。然后，它会摄取到 S3 存储桶中。

摄取工作正常，一切都很好，但是现在我需要在将Avro数据放入存储桶之前对其进行压缩。

我已经尝试了以下配置

   {
  "name":"--private-v1-s3-sink",
  "connector.class":"io.confluent.connect.s3.S3SinkConnector",
  "tasks.max": "1",
  "s3.region":"eu-west-1",
  "partition.duration.ms":"3600000",
  "rotate.schedule.interval.ms": "3600000",
  "topics.dir":"svs",
  "flush.size":"2500",
  "schema.compatibility":"FULL",
  "file.delim":"_",
  "topics":"--connect.s3.format.avro.AvroFormat",
  "key.converter":"org.apache.kafka.connect.storage.StringConverter",
  "value.converter":"io.confluent.connect.avro.AvroConverter",
  "value.converter.schema.registry.url":"--systems",
  "schema.generator.class":"io.confluent.connect.storage.hive.schema.DefaultSchemaGenerator",
  "partitioner.class":"io.confluent.connect.storage.partitioner.TimeBasedPartitioner",
  "storage.class":"io.confluent.connect.s3.storage.S3Storage",
  "s3.bucket.name": "${S3_BUCKET}",
  "s3.acl.canned":"bucket-owner-full-control",
  "avro.codec": "snappy",
  "locale":"en-GB",
  "timezone": "GMT",
  "errors.tolerance": "all",
  "path.format":"'ingest_date'=yyyy-MM-dd",
  "timestamp.extractor":"Record"

“avro.code”，我以为会压缩数据，但它没有。取而代之的是，我还尝试了“ ”s3.compression.type“： ”snappy“ '，仍然没有运气！但是，这确实适用于JSON和GZIP。

不太清楚哪里出了问题？

Answer 1

对于将来可能会遇到这种情况的人。

我在此设置之间运行了测试，使用 BZIP2 而不是 snappy 并且没有启用压缩。

结果是：

No compression 58.2MB / 406 total objects
BZIP    19.9MB / 406 total objects
Snappy 31.1MB / 406 total objects

在24小时的时间里，所有人都从同一个话题中提取并放入自己的桶中。

正如你所看到的，上面使用snappy的配置实际上是有效的。

BZIP提供了更高的压缩率，而且似乎更快。

最终，我们不得不使用snappy，因为Redshift摄取仅允许使用snappy压缩Avro，无论如何，此时。

Answer 2

这些设置仅适用于S3 Avro编写器，不适用于来自制作方的动态数据，这些数据必须在制作方或代理/主题级别进行“压缩”,而不是连接设置。

参考compression.type主题配置