如何捕捉 faust python json 序列化错误？答案

【问题标题】：How to catch faust python json serialization error?如何捕捉 faust python json 序列化错误？
【发布时间】：2021-07-23 19:26:50
【问题描述】：

我有简单的浮士德代理。它使用来自 kafka 主题的 json，并默认将它们解析为 dicts faust 序列化器：

@app.agent(source_topic, sink=[destination_topic])
async def fetch(records):
    async for record in records:
        result = do_some_stuff(record)
        yield result

反序列化本身发生在我的代码之外的某个地方，它由 faust 框架管理，而不是由我管理。如何捕获和处理反序列化异常，例如在无效 json 的情况下？

【问题讨论】：

您问了两个问题，一个是这个，另一个是stackoverflow.com/questions/67326259/…。也许用一个问题来简单地找出错误？除非两者不同
它们很相似，所以我刚刚删除了最后一个问题
我没有设置来测试它，但是有一个Open PR 表示没有为传递的参数调用on_error，但也有一个解决方法，如图所示here，也许你可以试试

标签： python json apache-kafka faust

【解决方案1】：

您可以手动序列化数据并捕获错误：

topic = app.topic('custom', value_type=bytes)

@app.agent
async def processor(stream):
    async for payload in stream:
        data = json.loads(payload)

来源：https://faust.readthedocs.io/en/latest/userguide/models.html

【讨论】：