如何将TopicPartition格式输出转换为指定结构的JSON Dump

你可以直接基于原始的偏移量字典（以TopicPartition对象为键）来转换，无需依赖那个丢失了偏移量信息的中间JSON。具体步骤如下：

1. 遍历原始字典的键值对，提取每个TopicPartition对象的partition属性，以及对应的偏移量值；
2. 将每一组[分区号, 偏移量]组成子列表，最终生成目标结构的数组；
3. 使用json.dumps()将数组转为JSON字符串。

代码示例 #

from kafka.structs import TopicPartition
import json

# 原始偏移量字典（从Kafka获取的结果）
offset_dict = {
    TopicPartition(topic='python-test', partition=0): 5,
    TopicPartition(topic='python-test', partition=1): 20,
    TopicPartition(topic='python-test', partition=2): 0
}

# 转换为目标结构
target_list = [[tp.partition, offset] for tp, offset in offset_dict.items()]

# 生成最终JSON
final_json = json.dumps(target_list)
print(final_json)

执行后输出：

[[0,5],[1,20],[2,0]]

如果只能基于你提到的中间JSON（仅包含topic和分区号）来处理，需要先将原始字典转为分区号到偏移量的映射，再通过分区号匹配组合数据：

备选代码示例 #

import json
from kafka.structs import TopicPartition

# 中间JSON字符串
intermediate_json = '[["python-test", 0], ["python-test", 1], ["python-test", 2]]'
intermediate_data = json.loads(intermediate_json)

# 原始偏移量字典
offset_dict = {
    TopicPartition(topic='python-test', partition=0): 5,
    TopicPartition(topic='python-test', partition=1): 20,
    TopicPartition(topic='python-test', partition=2): 0
}

# 构建分区号与偏移量的映射
partition_offset_map = {tp.partition: offset for tp, offset in offset_dict.items()}

# 生成目标结构
target_list = [[item[1], partition_offset_map[item[1]]] for item in intermediate_data]
final_json = json.dumps(target_list)
print(final_json)

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