"no root units" error in Dataflow, from PubSub to Bigquery in Golang-CodePudding

I am trying to read a message from PubSub, then write into BigQuery table in DataFlow. However, I faced "no root units" error by using direct runner.

Here is my code;

package main

import (
    "context"
    "encoding/json"
    "flag"
    "fmt"

    "github.com/apache/beam/sdks/v2/go/pkg/beam/io/bigqueryio"
    "github.com/apache/beam/sdks/v2/go/pkg/beam/x/debug"

    "github.com/apache/beam/sdks/v2/go/pkg/beam"
    "github.com/apache/beam/sdks/v2/go/pkg/beam/io/pubsubio"
    "github.com/apache/beam/sdks/v2/go/pkg/beam/log"
    "github.com/apache/beam/sdks/v2/go/pkg/beam/x/beamx"
)


type DummyBody struct {
        TaskId string `json:"id" bigquery:"id"`
    }


func buildPipeline(s beam.Scope) {
    rawDummyBodies := pubsubio.Read(s, "project", "topic", &pubsubio.ReadOptions{Subscription: "sub.ID"})

    dummyBodies := beam.ParDo(s, func(ctx context.Context, data []byte) (DummyBody, error) {
        var body DummyBody
        if err := json.Unmarshal(data, &body); err != nil {
            log.Error(ctx, err)
            fmt.Println("Error")
            return body, err
        }
        fmt.Println("No Error")
        return body, nil
    }, rawDummyBodies)

    debug.Printf(s, "Task : %#v", dummyBodies)

    bigqueryio.Write(s, "project", "table", dummyBodies)
}

func main() {
    flag.Parse()
    beam.Init()

    p, s := beam.NewPipelineWithRoot()
    buildPipeline(s)

    ctx := context.Background()
    if err := beamx.Run(ctx, p); err != nil {
        log.Exitf(ctx, "Failed to execute pipeline: %v", err)
    }
}

The pipeline started to execute with direct runner, but it failed due to no root units.

2022/11/01 14:29:55 Failed to execute pipeline: translation failed caused by: no root units exit status 1

CodePudding user response：

The current implementation of pubsubio only works on Dataflow Runner.