avro.go

avro.go

package main

import (
	"bytes"
	"fmt"
	"log"
	"os"
	"time"

	"github.com/linkedin/goavro"
)

func main() {
	// avro schema defintion
	codec, err := goavro.NewCodec(`
{
	"type": "record",
	"name": "service.location.gps",
	"doc": "device location by lat and long",
	"fields" : [
	{
		"name": "device",
		"type": "string"
	},
	{
		"name": "acqTime",
		"type": {
			"type": "int",
			"doc": "unix timestamp",
			"logicalType": "timestamp"
		}
	},
	{
		"name": "lat",
		"type": "double"
	},
	{
		"name": "lon",
		"type": "double"
	}
	]
}`)

	msg := make(map[string]interface{})
	// Define avro message content
	msg["device"] = "deviceID"
	msg["acqTime"] = time.Now().Unix()
	msg["lat"] = 1.6516
	msg["lon"] = 2.6516

	f := new(bytes.Buffer)
	if err != nil {
		log.Fatalf("Failed to create event.avro file: %v", err)
	}

	ocfw, err := goavro.NewOCFWriter(goavro.OCFConfig{
		W:     f,
		Codec: codec,
	})
	if err != nil {
		log.Fatalf("Failed to create the OCF Writer: %v", err)
	}

	err = ocfw.Append([]interface{}{msg})
	if err != nil {
		log.Fatalf("Failed to append to bin: %v", err)
	}

	// Reader
	ocfr, err := goavro.NewOCFReader(f)
	if err != nil {
		fmt.Println(err)
		os.Exit(1)
	}
	r := ocfr.Codec().Schema()
	readCodec, err := goavro.NewCodec(r)
	if err != nil {
		log.Fatalln(err)
	}
	m := make(map[string]interface{})
	for ocfr.Scan() {
		datum, _ := ocfr.Read()
		m = datum.(map[string]interface{})
	}

	// Output as direct access
	fmt.Println("Reading...")
	fmt.Println("device: " + m["device"].(string))
	fmt.Println("acqTime: " + fmt.Sprintf("%d", (m["acqTime"].(int32))))
	fmt.Println("lat: " + fmt.Sprintf("%f", (m["lat"].(float64))))
	fmt.Println("lon: " + fmt.Sprintf("%f", (m["lon"].(float64))))

	// Output as json
	jbytes, err := readCodec.TextualFromNative(nil, m)
	if err != nil {
		log.Fatalln(err)
	}
	fmt.Println(string(jbytes))
}