如何使用 Golang 函数处理流数据？

package main

import (
    "context"
    "fmt"
    "math/big"
    "strconv"
    "time"

    "cloud.google.com/go/bigquery"
)

func main() {
    ctx := context.Background()
    client, err := bigquery.NewClient(ctx, "project-id")
    if err != nil {
        // Handle error.
    }

    // 设置源表
    sourceTable := client.Dataset("my_dataset").Table("source_table")

    // 设置目标表
    destinationTable := client.Dataset("my_dataset").Table("destination_table")
    if err := destinationTable.Delete(ctx); err != nil {
        // Handle error.
    }

    // 创建查询
    query := client.Query(
        `SELECT
            Name,
            COUNT(*) AS count
        FROM ${sourceTable}
        GROUP BY Name`)

    // Create the UDF definition
    avgLengthUDF := bigquery.UDFFromFunction(func(ctx context.Context, rows []bigquery.Value) (bigquery.Value, error) {
        var total big.Int

        for _, row := range rows {
            // 从每一行中获取长度。
            length, ok := row[1].Value().(int64)
            if !ok {
                return nil, fmt.Errorf("invalid length '%v'", row[1].Value())
            }
            total.Add(&total, big.NewInt(length))
        }

        // 计算平均长度。
        avg := new(big.Float).Quo(new(big.Float).SetInt64(total.Int64()), new(big.Float).SetInt64(len(rows)))

        // 返回平均长度作为结果。
        return avg.String(), nil
    })

    // 更新查询以包含 UDF
    query.UDFs = map[string]bigquery.ScalarUDF{
        "avg_length": avgLengthUDF,
    }

    job, err := query.Run(ctx)
    if err != nil {
        // Handle error.
    }
    status, err := job.Wait(ctx)
    if err != nil {
        // Handle error.
    }

    if status.Err() != nil {
        // Handle error.
    }

    it := destinationTable.Read(ctx)
    for {
        var row []bigquery.Value
        err := it.Next(&row)
        if err == iterator.Done {
            break
        }
        if err != nil {
            // Handle error.
        }

        fmt.Println(row)
    }
}