Avro 中未处理的循环引用答案

【问题标题】：Circular references not handled in AvroAvro 中未处理的循环引用
【发布时间】：2014-06-09 23:27:27
【问题描述】：

Avro 附带了一个名为 Avro-Tools 的工具，可用于在 JSON、Avro-Schema (.avsc) 和二进制格式之间进行转换。但它不适用于循环引用。

我们有两个文件：

circular.avsc（由 Avro 生成）
circular.json（由 Jackson 生成，因为它具有循环引用，而 Avro 不喜欢它）。

circle.avsc

{
   "type":"record",
   "name":"Parent",
   "namespace":"bigdata.example.avro",
   "fields":[
      {
         "name":"name",
         "type":[
            "null",
            "string"
         ],
         "default":null
      },
      {
         "name":"child",
         "type":[
            "null",
            {
               "type":"record",
               "name":"Child",
               "fields":[
                  {
                     "name":"name",
                     "type":[
                        "null",
                        "string"
                     ],
                     "default":null
                  },
                  {
                     "name":"parent",
                     "type":[
                        "null",
                        "Parent"
                     ],
                     "default":null
                  }
               ]
            }
         ],
         "default":null
      }
   ]
}

循环.json

{
   "@class":"bigdata.example.avro.Parent",
   "@circle_ref_id":1,
   "name":"parent",
   "child":{
      "@class":"bigdata.example.avro.DerivedChild",
      "@circle_ref_id":2,
      "name":"hello",
      "parent":1
   }
}

在上面运行avro-tools的命令

java -jar avro-tools-1.7.6.jar fromjson --schema-file circular.avsc circular.json

输出

2014-06-09 14:29:17.759 java[55860:1607] 无法从 SCDynamicStore 加载领域映射信息 Objavro.codenullavro.schema？ {"type":"record","name":"Parent","namespace":"bigdata.example.avro","fields":[{"name":"name","type":["null ","string"],"default":null},{"name":"child","type":["null",{"type":"record","name":"Child","字段":[{"name":"name","type":["null","string"],"default":null},{"name":"parent","type":["null ","Parent"],"default":null}]}],"default":null}]}?'???K?jH!??Ė? 线程“main”org.apache.avro 中的异常。 AvroTypeException：预期的启动联合。收到 VALUE_STRING 在 org.apache.avro.io.JsonDecoder.error(JsonDecoder.java:697)

在 org.apache.avro.io.JsonDecoder.readIndex(JsonDecoder.java:441)

在 org.apache.avro.io.ResolvingDecoder.doAction(ResolvingDecoder.java:229)

其他一些 JSON 值尝试使用相同的架构，但没有成功

JSON 1

{
   "name":"parent",
   "child":{
      "name":"hello",
      "parent":null
   }
}

JSON 2

{
   "name":"parent",
   "child":{
      "name":"hello",
   }
}

JSON 3

 {
   "@class":"bigdata.example.avro.Parent",
   "@circle_ref_id":1,
   "name":"parent",
   "child":{
      "@class":"bigdata.example.avro.DerivedChild",
      "@circle_ref_id":2,
      "name":"hello",
      "parent":null
   }
}

删除一些“可选”元素：

circle.avsc

{
   "type":"record",
   "name":"Parent",
   "namespace":"bigdata.example.avro",
   "fields":[
      {
         "name":"name",
         "type":
            "string",
         "default":null
      },
      {
         "name":"child",
         "type":
            {
               "type":"record",
               "name":"Child",
               "fields":[
                  {
                 "name":"name",
                 "type":
                    "string",
                 "default":null
                  },
                  {
                     "name":"parent",
                     "type":
                        "Parent",
                     "default":null
                  }
               ]
            },
         "default":null
      }
   ]
}

circular.json

 {
   "@class":"bigdata.example.avro.Parent",
   "@circle_ref_id":1,
   "name":"parent",
   "child":{
      "@class":"bigdata.example.avro.DerivedChild",
      "@circle_ref_id":2,
      "name":"hello",
      "parent":1
   }
}

输出

2014-06-09 15:30:53.716 java[56261:1607] 无法从 SCDynamicStore 加载领域映射信息 Objavro.codenullavro.schema?{"type":"record","name":"Parent","namespace":"bigdata.example.avro","fields":[{"name":"name"," type":"string","default":null},{"name":"child","type":{"type":"record","name":"Child","fields":[{ "name":"name","type":"string","default":null},{"name":"parent","type":"Parent","default":null}]}," default":null}]}?x?N??O"?M?`AbException in thread "main" java.lang.StackOverflowError

在 org.apache.avro.io.parsing.Symbol.flattenedSize(Symbol.java:212)

在 org.apache.avro.io.parsing.Symbol$Sequence.flattenedSize(Symbol.java:323)

在 org.apache.avro.io.parsing.Symbol.flattenedSize(Symbol.java:216)

在 org.apache.avro.io.parsing.Symbol$Sequence.flattenedSize(Symbol.java:323)

在 org.apache.avro.io.parsing.Symbol.flattenedSize(Symbol.java:216)

在 org.apache.avro.io.parsing.Symbol$Sequence.flattenedSize(Symbol.java:323)

有人知道我如何使用 Avro 进行循环引用吗？

【问题讨论】：

标签： json apache hadoop jackson avro

【解决方案1】：

我最近遇到了同样的问题，并以一种变通的方式解决了，希望它可以帮助。

基于Avro specification：

JSON 编码除了联合之外，JSON 编码与用于对字段默认值进行编码相同。

联合的值在 JSON 中编码如下：

如果其类型为null，则编码为JSON null；
否则，它将被编码为具有一个名称/值对的 JSON 对象，其名称是类型的名称，其值是递归编码的值。对于 Avro 的命名类型（记录、固定或枚举），使用用户指定的名称，对于其他类型，使用类型名称。

例如，联合模式 ["null","string","Foo"]，其中 Foo 是记录名称，将编码：

null 为空；
字符串 "a" 为 {"string": "a"};
Foo 实例为 {"Foo": {...}}，其中 {...} 表示 Foo 实例的 JSON 编码。

如果无法更改源文件以符合要求，也许我们必须更改代码。所以我从 avro-1.7.7 包中定制了原始的 org.apache.avro.io.JsonDecoder 类，并创建了我自己的类 MyJsonDecoder。

除了创建新的构造函数和类名之外，这是我更改的关键：

    @Override
public int readIndex() throws IOException {
    advance(Symbol.UNION);
    Symbol.Alternative a = (Symbol.Alternative) parser.popSymbol();

    String label;
    if (in.getCurrentToken() == JsonToken.VALUE_NULL) {
        label = "null";
//***********************************************
// Original code: according to Avor document "JSON Encoding":
// it is encoded as a Json object with one name/value pair whose name is
//   the type's name and whose value is the recursively encoded value.
// Can't change source data, so remove this rule.
//        } else if (in.getCurrentToken() == JsonToken.START_OBJECT &&
//                in.nextToken() == JsonToken.FIELD_NAME) {
//            label = in.getText();
//            in.nextToken();
//            parser.pushSymbol(Symbol.UNION_END);
//***********************************************
        // Customized code:
        // Add to check if type is in the union then parse it.
        // Check if type match types in union or not.
    } else {
        label = findTypeInUnion(in.getCurrentToken(), a);

        // Field missing but not allow to be null
        //   or field type is not in union.
        if (label == null) {
            throw error("start-union, type may not be in UNION,");
        }
    }
//***********************************************
// Original code: directly error out if union
//        } else {
//                throw error("start-union");
//        }
//***********************************************
    int n = a.findLabel(label);
    if (n < 0)
        throw new AvroTypeException("Unknown union branch " + label);
    parser.pushSymbol(a.getSymbol(n));
    return n;
}

/**
 * Method to check if current JSON token type is declared in union.
 * Do NOT support "record", "enum", "fix":
 * Because there types require user defined name in Avro schema,
 * if user defined names could not be found in Json file, can't decode.
 *
 * @param jsonToken         JsonToken
 * @param symbolAlternative Symbol.Alternative
 * @return String Parsing label, decode in which way.
 */
private String findTypeInUnion(final JsonToken jsonToken,
                               final Symbol.Alternative symbolAlternative) {
    // Create a map for looking up: JsonToken and Avro type
    final HashMap<JsonToken, String> json2Avro = new HashMap<>();

    for (int i = 0; i < symbolAlternative.size(); i++) {
        // Get the type declared in union: symbolAlternative.getLabel(i).
        // Map the JsonToken with Avro type.
        switch (symbolAlternative.getLabel(i)) {
            case "null":
                json2Avro.put(JsonToken.VALUE_NULL, "null");
                break;
            case "boolean":
                json2Avro.put(JsonToken.VALUE_TRUE, "boolean");
                json2Avro.put(JsonToken.VALUE_FALSE, "boolean");
                break;
            case "int":
                json2Avro.put(JsonToken.VALUE_NUMBER_INT, "int");
                break;
            case "long":
                json2Avro.put(JsonToken.VALUE_NUMBER_INT, "long");
                break;
            case "float":
                json2Avro.put(JsonToken.VALUE_NUMBER_FLOAT, "float");
                break;
            case "double":
                json2Avro.put(JsonToken.VALUE_NUMBER_FLOAT, "double");
                break;
            case "bytes":
                json2Avro.put(JsonToken.VALUE_STRING, "bytes");
                break;
            case "string":
                json2Avro.put(JsonToken.VALUE_STRING, "string");
                break;
            case "array":
                json2Avro.put(JsonToken.START_ARRAY, "array");
                break;
            case "map":
                json2Avro.put(JsonToken.START_OBJECT, "map");
                break;
            default: break;
        }
    }

    // Looking up the map to find out related Avro type to JsonToken
    return json2Avro.get(jsonToken);
}

生成思路是检查源文件中的类型是否可以在联合中找到。

这里还有一些问题：

此解决方案不支持“记录”、“枚举”或“固定”Avro 类型，因为这些类型需要用户定义的名称。例如。如果你想要联合 "type": ["null", {"name": "abc", "type": "record", "fields" : ...}]，这段代码将不起作用。对于原始类型，这应该有效。但请在将其用于您的项目之前对其进行测试。
我个人认为记录不应该为空，因为我认为记录是我需要确保存在的东西，如果缺少某些东西，这意味着我有更大的问题。如果可以省略，我更喜欢在定义架构时使用“map”作为类型，而不是使用“record”。

希望这会有所帮助。

【讨论】：

有一个 avro ticket 用于处理这些循环引用。 (issues.apache.org/jira/browse/AVRO-695) 不知道是什么阻碍了它。也许没有足够多的人在寻找这样的功能，它需要更多的投票才能让它对提交足够重要。