Flink广播状态不支持schemaevolution

为了避免Flink广播状态不支持schema evolution的问题，可以使用一个转换函数将原始数据转换为新的数据格式。然后，在应用程序中使用这个转换函数，以便将状态中的旧数据转换为新的数据格式。

下面的代码示例展示了如何使用转换函数来解决广播状态不支持schema evolution的问题。假设有一个广播状态包含了以下数据：

{"id": 1, "name": "Alice", "age": 30}
{"id": 2, "name": "Bob", "age": 25}

但现在需要将该状态的数据格式更改为：

{"id": 1, "first_name": "Alice", "last_name": "Smith"}
{"id": 2, "first_name": "Bob", "last_name": "Johnson"}

为了使状态能够支持schema evolution，我们需要编写一个将旧数据转换为新数据格式的函数。示例代码如下：

public static class PersonConverter implements MapFunction<Tuple2<Integer, String>, Tuple2<Integer, Person>> {
    @Override
    public Tuple2<Integer, Person> map(Tuple2<Integer, String> value) throws Exception {
        String json = value.f1;
        ObjectMapper mapper = new ObjectMapper();
        JsonNode node = mapper.readTree(json);
        int id = node.get("id").asInt();
        String name = node.get("name").asText();
        int age = node.get("age").asInt();

        String first_name = name.split(" ")[0];
        String last_name = name.split(" ")[1];
        Person person = new Person(id, first_name, last_name, age);
        return new Tuple2<>(id, person);
    }
}

public static class Person implements Serializable {
    public int id;
    public String first_name;
    public String last_name;
    public int age;

    public Person() {
    }

    public Person(int id, String first_name, String last_name, int age) {
        this.id = id;
        this.first_name = first_name;
        this.last