我想做什么?
对于数据库中的每一次更改,我都试图将 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
可用于将接口(interface)转换为字符串。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)
}
关于go - 使用 goavro 处理多种类型的解码数据,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/63454030/