使用 goavro 处理多种类型的解码数据

从现在开始，我们要努力学习啦！今天我给大家带来《使用 goavro 处理多种类型的解码数据》，感兴趣的朋友请继续看下去吧！下文中的内容我们主要会涉及到等等知识点，如果在阅读本文过程中有遇到不清楚的地方，欢迎留言呀！我们一起讨论，一起学习！

我想做什么？

对于数据库中的每项更改，我都尝试将 debezium 事件转换为数据库值的 csv，以便加载到 redshift。

对于以下 110 更改，我正在尝试制作一个 csv 文件：110,vck,desc,221.1

mysql> select * from products;
+-----+-------------+---------------------------------------------------------+--------+
| id  | name        | description                                             | weight |
+-----+-------------+---------------------------------------------------------+--------+
| 110 | vck         | desc                                                    |  221.1 |
+-----+-------------+---------------------------------------------------------+--------+

这是我尝试使用 goavro 来做到这一点。

程序

https://play.golang.org/p/a8wd0szpueq

package main

import (
    "fmt"
    "encoding/json"
)

func main() {
    debeziumEvent := `{"before":null,"after":{"datapipe.inventory.products.Value":{"id":110,"name":"vck","description":{"string":"desc"},"weight":{"double":221.10000610351562}}},"source":{"query":null,"snapshot":{"string":"true"},"server_id":0,"gtid":null,"name":"datapipe","thread":null,"ts_ms":0,"file":"mysql-bin.000049","version":"1.2.1.Final","connector":"mysql","pos":154,"table":{"string":"products"},"row":0,"db":"inventory"},"op":"c","ts_ms":{"long":1597649700266},"transaction":null}`
    
    var data map[string]interface{}
        err := json.Unmarshal([]byte(debeziumEvent), &data)
        if err != nil {
            panic(err)
        }
    
    after := data["after"].(map[string]interface{})
    csv := make([]interface{}, 0)
    
    for _, v := range after {
        for _, v2 := range v.(map[string]interface{}) {
            switch stype := v2.(type) {
            case map[string]interface{}:
                for _, v3 := range v2.(map[string]interface{}) {
                    csv = append(csv, v3)
                }
            case string:
                csv = append(csv, v2)
            case int:
                csv = append(csv, v2)
            case float64:
                csv = append(csv, v2)
            default:
                fmt.Printf("type %s not handled\n", stype) 
                panic("unhandled type")
            }
            
        }
    }
    
    fmt.Println(csv)
}

有没有更好的方法？对于每种数据类型，我都需要在这里有一个 switch 语句......

链接的 goavro 问题：https://github.com/linkedin/goavro/issues/217

解决方案

fmt.sprintf 可用于将接口转换为字符串。 str := fmt.sprintf("%v", v)

这样做可以将 case 语句减少到 2 个：

package main

import (
    "fmt"
    "encoding/json"
)

func main() {
    debeziumEvent := `{"before":null,"after":{"datapipe.inventory.products.Value":{"id":110,"name":"vck","description":{"string":"desc"},"weight":{"double":221.10000610351562}}},"source":{"query":null,"snapshot":{"string":"true"},"server_id":0,"gtid":null,"name":"datapipe","thread":null,"ts_ms":0,"file":"mysql-bin.000049","version":"1.2.1.Final","connector":"mysql","pos":154,"table":{"string":"products"},"row":0,"db":"inventory"},"op":"c","ts_ms":{"long":1597649700266},"transaction":null}`
    
    var data map[string]interface{}
        err := json.Unmarshal([]byte(debeziumEvent), &data)
        if err != nil {
            panic(err)
        }
    //fmt.Printf("data=%v\n", data)
    
    after := data["after"].(map[string]interface{})
    csv := []string{}
    
    for _, v := range after {
        for _, v2 := range v.(map[string]interface{}) {
            switch v2.(type) {
            case map[string]interface{}:
                for _, v3 := range v2.(map[string]interface{}) {
                    csv = append(csv, fmt.Sprintf("%v", v3))
                }
            default:
                csv = append(csv, fmt.Sprintf("%v", v2))
            }
            
        }
    }
    
    fmt.Println(csv)
}

终于介绍完啦！小伙伴们，这篇关于《使用 goavro 处理多种类型的解码数据》的介绍应该让你收获多多了吧！欢迎大家收藏或分享给更多需要学习的朋友吧~golang学习网公众号也会发布Golang相关知识，快来关注吧！