go - 将debezium事件模式解码为golang中有意义的数据结构

Question

Debezium模式是一个字符串(json):

{
   "type":"record",
   "name":"Envelope",
   "namespace":"datapipe.inventory.customers",
   "fields":[
      {
         "name":"before",
         "type":[
            "null",
            {
               "type":"record",
               "name":"Value",
               "fields":[
                  {
                     "name":"id",
                     "type":"int"
                  },
                  {
                     "name":"first_name",
                     "type":"string"
                  },
                  {
                     "name":"last_name",
                     "type":"string"
                  },
                  {
                     "name":"email",
                     "type":"string"
                  }
               ],
               "connect.name":"datapipe.inventory.customers.Value"
            }
         ],
         "default":null
      },
      {
         "name":"after",
         "type":[
            "null",
            "Value"
         ],
         "default":null
      },
      {
         "name":"source",
         "type":{
            "type":"record",
            "name":"Source",
            "namespace":"io.debezium.connector.mysql",
            "fields":[
               {
                  "name":"version",
                  "type":"string"
               },
               {
                  "name":"connector",
                  "type":"string"
               },
               {
                  "name":"name",
                  "type":"string"
               },
               {
                  "name":"ts_ms",
                  "type":"long"
               },
               {
                  "name":"snapshot",
                  "type":[
                     {
                        "type":"string",
                        "connect.version":1,
                        "connect.parameters":{
                           "allowed":"true,last,false"
                        },
                        "connect.default":"false",
                        "connect.name":"io.debezium.data.Enum"
                     },
                     "null"
                  ],
                  "default":"false"
               },
               {
                  "name":"db",
                  "type":"string"
               },
               {
                  "name":"table",
                  "type":[
                     "null",
                     "string"
                  ],
                  "default":null
               },
               {
                  "name":"server_id",
                  "type":"long"
               },
               {
                  "name":"gtid",
                  "type":[
                     "null",
                     "string"
                  ],
                  "default":null
               },
               {
                  "name":"file",
                  "type":"string"
               },
               {
                  "name":"pos",
                  "type":"long"
               },
               {
                  "name":"row",
                  "type":"int"
               },
               {
                  "name":"thread",
                  "type":[
                     "null",
                     "long"
                  ],
                  "default":null
               },
               {
                  "name":"query",
                  "type":[
                     "null",
                     "string"
                  ],
                  "default":null
               }
            ],
            "connect.name":"io.debezium.connector.mysql.Source"
         }
      },
      {
         "name":"op",
         "type":"string"
      },
      {
         "name":"ts_ms",
         "type":[
            "null",
            "long"
         ],
         "default":null
      },
      {
         "name":"transaction",
         "type":[
            "null",
            {
               "type":"record",
               "name":"ConnectDefault",
               "namespace":"io.confluent.connect.avro",
               "fields":[
                  {
                     "name":"id",
                     "type":"string"
                  },
                  {
                     "name":"total_order",
                     "type":"long"
                  },
                  {
                     "name":"data_collection_order",
                     "type":"long"
                  }
               ]
            }
         ],
         "default":null
      }
   ],
   "connect.name":"datapipe.inventory.customers.Envelope"
}

我想将此模式解码为有意义的数据结构，以便可以对该数据执行操作。
使用map和for循环进行转换的速度很慢，并且需要进行如下类似的操作。

for k1, v2 := range m["fields"].([]interface{}) {
        fmt.Println(k1)
        v3 := v2.(map[string]interface{})
        for _, value := range v3 {
            if value == "before" {
                vtype := v3["type"].([]interface{})

                for kk, vtyp := range vtype {
                    fmt.Printf("kk %v, vtyp %v\n", kk, vtyp)
                    // for k4, v4 := range vtyp.([]interface{}) {
                    //  fmt.Printf("key4: %v, value4: %v\n", k4, v4)
                    // }

                }


            }
        }
    }

Golang是否提供Debezium类型？请提出建议。我确实检查了文档，但是使用太多嵌套的类型非常令人困惑。
相关 https://github.com/riferrei/srclient/issues/13

Answer 1

package main

import (
    "fmt"
    "encoding/json"
)

type DebeziumSchema struct {
    Type        string         `json:"type"`
    Name        string         `json:"name"`
    Namespace   string         `json:"namespace"`
    Fields      []SchemaField  `json:"fields"`
    ConnectName string         `json:"connect.name"`
}

type SchemaField struct {
    Name    string          `json:"name"`
    Type    interface{}     `json:"type"`
    Default interface{}     `json:"default,omitempty"`
}

func main() {
    schema := `{"type":"record","name":"Envelope","namespace":"datapipe.inventory.customers","fields":[{"name":"before","type":["null",{"type":"record","name":"Value","fields":[{"name":"id","type":"int"},{"name":"first_name","type":"string"},{"name":"last_name","type":"string"},{"name":"email","type":"string"}],"connect.name":"datapipe.inventory.customers.Value"}],"default":null},{"name":"after","type":["null","Value"],"default":null},{"name":"source","type":{"type":"record","name":"Source","namespace":"io.debezium.connector.mysql","fields":[{"name":"version","type":"string"},{"name":"connector","type":"string"},{"name":"name","type":"string"},{"name":"ts_ms","type":"long"},{"name":"snapshot","type":[{"type":"string","connect.version":1,"connect.parameters":{"allowed":"true,last,false"},"connect.default":"false","connect.name":"io.debezium.data.Enum"},"null"],"default":"false"},{"name":"db","type":"string"},{"name":"table","type":["null","string"],"default":null},{"name":"server_id","type":"long"},{"name":"gtid","type":["null","string"],"default":null},{"name":"file","type":"string"},{"name":"pos","type":"long"},{"name":"row","type":"int"},{"name":"thread","type":["null","long"],"default":null},{"name":"query","type":["null","string"],"default":null}],"connect.name":"io.debezium.connector.mysql.Source"}},{"name":"op","type":"string"},{"name":"ts_ms","type":["null","long"],"default":null},{"name":"transaction","type":["null",{"type":"record","name":"ConnectDefault","namespace":"io.confluent.connect.avro","fields":[{"name":"id","type":"string"},{"name":"total_order","type":"long"},{"name":"data_collection_order","type":"long"}]}],"default":null}],"connect.name":"datapipe.inventory.customers.Envelope"}`

    var d DebeziumSchema

    err := json.Unmarshal([]byte(schema), &d)
    if err != nil {
        panic(err)
    }

    columns := make(map[string]string)
    
    for _, field := range d.Fields {
        if field.Name != "before" {
            continue
        }
        
        for _, k := range field.Type.([]interface{}) {
            switch k.(type) {
            case map[string]interface{}:
                rk := k.(map[string]interface{})
                for rk1, rk2 := range rk {
                    if rk1 != "fields" {
                        continue
                    }
                    for _, kkl := range rk2.([]interface{}) {
                        k22 := kkl.(map[string]interface{})
                        columns[k22["name"].(string)] = k22["type"].(string)
                    }
                }
                
            }
        }
        
    }
    
    fmt.Println(columns)

}

map[email:string first_name:string id:int last_name:string]

我能够提取这样做的列。但是我相信可以用更好的方法做到这一点。
https://play.golang.org/p/e0-F4Un4Mzv