英文:
"no root units" error in Dataflow, from PubSub to Bigquery in Golang
问题
我正在尝试从PubSub读取消息，然后在DataFlow中将其写入BigQuery表格。然而，当我使用直接运行器时，遇到了“no root units”错误。
这是我的代码：
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)
}
}
该流水线开始使用直接运行器执行，但由于没有根单元，执行失败。
2022/11/01 14:29:55 Failed to execute pipeline: translation failed
caused by:
no root units
exit status 1
英文:
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
答案1
得分: 6
pubsubio的当前实现仅适用于Dataflow Runner。
英文:
The current implementation of pubsubio only works on Dataflow Runner.
通过集体智慧和协作来改善编程学习和解决问题的方式。致力于成为全球开发者共同参与的知识库，让每个人都能够通过互相帮助和分享经验来进步。
评论