Flink CDC里cdc mongo 碰到这样的数据咋处理？

在Flink CDC中，对于MongoDB这样的非结构化数据源，我们通常使用JSON解析器来处理数据。对于你提到的这种数据"freightrates":{"$numberLong":"100000"}，我们可以将其转换为Java的Number类型，然后进行后续的处理。

以下是一个示例代码：

import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.bson.Document;

public class MongoDBDeserializationSchema {
    private static final TypeInformation<Document> DOCUMENT_TYPE_INFO = TypeExtractor.getForClass(Document.class);

    public static Document deserialize(String json) {
        // 使用JSON解析器将JSON字符串转换为Document对象
        Document document = Document.parse(json);

        // 遍历Document对象，将"$numberLong"类型的值转换为Number类型
        for (Map.Entry<String, Object> entry : document.entrySet()) {
            Object value = entry.getValue();
            if (value instanceof String && value.toString().startsWith("$numberLong:")) {
                String numberLongStr = value.toString().substring("$numberLong:".length());
                long numberLong = Long.parseLong(numberLongStr);
                document.put(entry.getKey(), numberLong);
            }
        }

        return document;
    }
}

在这个示例中，我们首先定义了一个MongoDBDeserializationSchema类，其中包含一个静态方法deserialize，用于将JSON字符串转换为Document对象。然后，我们遍历Document对象，将"$numberLong"类型的值转换为Number类型。最后，返回处理后的Document对象。

这样BsonDocment.parse and then toStrinig。此回答来自钉群Flink CDC 社区。