我正在使用 Benthos 從 Kafka 讀取 AVRO 編碼的消息，其中kafka_key元數據字段設置為還包含 AVRO 編碼的有效負載。這些 AVRO 編碼的有效載荷的模式存儲在模式注冊表中，Benthos 有一個schema_registry_decode用于解碼它們的處理器。我希望為每個包含兩個字段的 Kafka 消息生成輸出 JSON 消息，一個稱為content包含解碼的 AVRO 消息，另一個稱為包含Benthos 收集metadata的各種元數據字段kafka_key，包括解碼的有效負載。
1 回答
隔江千里
TA貢獻1906條經驗 獲得超10個贊
事實證明，可以使用這樣的branch處理器來實現這一點：
input:
kafka:
addresses:
- localhost:9092
consumer_group: benthos_consumer_group
topics:
- benthos_input
pipeline:
processors:
# Decode the message
- schema_registry_decode:
url: http://localhost:8081
# Populate output content field
- bloblang: |
root.content = this
# Decode kafka_key metadata payload and populate output metadata field
- branch:
request_map: |
root = meta("kafka_key")
processors:
- schema_registry_decode:
url: http://localhost:8081
result_map: |
root.metadata = meta()
root.metadata.kafka_key = this
output:
stdout: {}
- 1 回答
- 0 關注
- 213 瀏覽
添加回答
舉報
0/150
提交
取消