python之每个 kafka 主题的 spark streaming 不同值解码器

我需要创建一个从多个主题读取的 Spark 流，并为每个主题使用不同的解码器(每个主题包含不同的 avro 编码对象):

def decode_avro(message): 
    schem = avro.schema.parse(open("error_list.avsc").read()) 
    bytes_reader = io.BytesIO(message) 
    decoder = avro.io.BinaryDecoder(bytes_reader) 
    reader = avro.io.DatumReader(schem) 
    return reader.read(decoder) 
 
ssc = StreamingContext(sc, 2) 
kvs = KafkaUtils.createDirectStream(ssc, [topic, topic2], { 
    "metadata.broker.list": brokers}, valueDecoder = decode_avro) 

我不想知道是否可以为每个主题指定不同的解码器回调，或者是否可以知道解码器函数上的主题名称(这样我可以使用 avro 模式文件的主题名称并在同一函数中解码所有消息)

谢谢

请您参考如下方法:

我们也有这样的情况，我们从不同的主题中读取不同的消息格式，然后处理每个主题并将输出存储到每个源主题的专用存储中。 正确的方法是创建多个流。在同一应用程序中使用相同的 Spark 上下文按主题流式传输。 每个流都将获得相关的 ValueDecoder，如果它们共享相同的格式，您仍然可以阅读多个主题。