我正在使用 Benthos 從 Kafka 讀取 AVRO 編碼的消息，其中kafka_key元數(shù)據(jù)字段設(shè)置為還包含 AVRO 編碼的有效負(fù)載。這些 AVRO 編碼的有效載荷的模式存儲(chǔ)在模式注冊(cè)表中，Benthos 有一個(gè)schema_registry_decode用于解碼它們的處理器。我希望為每個(gè)包含兩個(gè)字段的 Kafka 消息生成輸出 JSON 消息，一個(gè)稱為content包含解碼的 AVRO 消息，另一個(gè)稱為包含Benthos 收集metadata的各種元數(shù)據(jù)字段kafka_key，包括解碼的有效負(fù)載。
1 回答
隔江千里
TA貢獻(xiàn)1906條經(jīng)驗(yàn) 獲得超10個(gè)贊
事實(shí)證明，可以使用這樣的branch處理器來實(shí)現(xiàn)這一點(diǎn)：
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 關(guān)注
- 220 瀏覽
添加回答
舉報(bào)
0/150
提交
取消