如何配置Kafka S3 Sink Connector的FieldPartitioner实现多层分区？

Hey Julia, let's get this nested partitioning sorted out for you!

The main adjustment you need is to specify multiple fields in partition.field.name (comma-separated, no spaces) and set the partition delimiter to a slash to create the nested directory structure you want (field1=value/field2=value/field3=value). Here's the complete corrected config section:

# Your existing core configuration
storage.class=io.confluent.connect.s3.storage.S3Storage
format.class=io.confluent.connect.s3.format.parquet.ParquetFormat
schema.generator.class=io.confluent.connect.storage.hive.schema.DefaultSchemaGenerator
schema.compatibility=NONE

# Updated partitioner configs
partitioner.class=io.confluent.connect.storage.partitioner.FieldPartitioner
# List your target fields separated by commas (exact case matches your message fields!)
partition.field.name=field1,field2,field3
# Use slash as delimiter to create nested paths
partition.delim=/
# Explicitly enable field names in partition paths (default is true, but safe to set explicitly)
partitioner.include.field.name=true

Key Details & Troubleshooting Tips: #

• Field Name Matching: Make sure the fields in partition.field.name exactly match the case and structure of your Kafka messages. For nested fields (like user.address.city in JSON payloads), use dot notation to reference them.
• Converter Setup: If you're using structured message formats (like JSON), ensure your converter is correctly configured. For example, if using schemaless JSON:

  value.converter=org.apache.kafka.connect.json.JsonConverter
  value.converter.schemas.enable=false
  

• Case Sensitivity: Kafka Connect is case-sensitive when looking up fields, so Field1 is not the same as field1—double-check your message payload structure to confirm exact field names.

内容的提问来源于stack exchange，提问作者Julia Bel