在Golang中，从PubSub到BigQuery的Dataflow中出现了"no root units"错误。

huangapple
go评论117阅读模式
英文:

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

问题

我正在尝试从PubSub读取消息，然后在DataFlow中将其写入BigQuery表格。然而，当我使用直接运行器时，遇到了“no root units”错误。

这是我的代码：

  1. package main
  3. import (
  4. 	"context"
  5. 	"encoding/json"
  6. 	"flag"
  7. 	"fmt"
  9. 	"github.com/apache/beam/sdks/v2/go/pkg/beam/io/bigqueryio"
  10. 	"github.com/apache/beam/sdks/v2/go/pkg/beam/x/debug"
  12. 	"github.com/apache/beam/sdks/v2/go/pkg/beam"
  13. 	"github.com/apache/beam/sdks/v2/go/pkg/beam/io/pubsubio"
  14. 	"github.com/apache/beam/sdks/v2/go/pkg/beam/log"
  15. 	"github.com/apache/beam/sdks/v2/go/pkg/beam/x/beamx"
  16. )
  18. type DummyBody struct {
  19. 	TaskId string `json:"id" bigquery:"id"`
  20. }
  22. func buildPipeline(s beam.Scope) {
  23. 	rawDummyBodies := pubsubio.Read(s, "project", "topic", &pubsubio.ReadOptions{Subscription: "sub.ID"})
  25. 	dummyBodies := beam.ParDo(s, func(ctx context.Context, data []byte) (DummyBody, error) {
  26. 		var body DummyBody
  27. 		if err := json.Unmarshal(data, &body); err != nil {
  28. 			log.Error(ctx, err)
  29. 			fmt.Println("Error")
  30. 			return body, err
  31. 		}
  32. 		fmt.Println("No Error")
  33. 		return body, nil
  34. 	}, rawDummyBodies)
  36. 	debug.Printf(s, "Task : %#v", dummyBodies)
  38. 	bigqueryio.Write(s, "project", "table", dummyBodies)
  39. }
  41. func main() {
  42. 	flag.Parse()
  43. 	beam.Init()
  45. 	p, s := beam.NewPipelineWithRoot()
  46. 	buildPipeline(s)
  48. 	ctx := context.Background()
  49. 	if err := beamx.Run(ctx, p); err != nil {
  50. 		log.Exitf(ctx, "Failed to execute pipeline: %v", err)
  51. 	}
  52. }

该流水线开始使用直接运行器执行，但由于没有根单元，执行失败。

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;

  1. package main
  2. import (
  3. "context"
  4. "encoding/json"
  5. "flag"
  6. "fmt"
  7. "github.com/apache/beam/sdks/v2/go/pkg/beam/io/bigqueryio"
  8. "github.com/apache/beam/sdks/v2/go/pkg/beam/x/debug"
  9. "github.com/apache/beam/sdks/v2/go/pkg/beam"
  10. "github.com/apache/beam/sdks/v2/go/pkg/beam/io/pubsubio"
  11. "github.com/apache/beam/sdks/v2/go/pkg/beam/log"
  12. "github.com/apache/beam/sdks/v2/go/pkg/beam/x/beamx"
  13. )
  14. type DummyBody struct {
  15. TaskId string `json:"id" bigquery:"id"`
  16. }
  17. func buildPipeline(s beam.Scope) {
  18. rawDummyBodies := pubsubio.Read(s, "project", "topic", &pubsubio.ReadOptions{Subscription: "sub.ID"})
  19. dummyBodies := beam.ParDo(s, func(ctx context.Context, data []byte) (DummyBody, error) {
  20. var body DummyBody
  21. if err := json.Unmarshal(data, &body); err != nil {
  22. log.Error(ctx, err)
  23. fmt.Println("Error")
  24. return body, err
  25. }
  26. fmt.Println("No Error")
  27. return body, nil
  28. }, rawDummyBodies)
  29. debug.Printf(s, "Task : %#v", dummyBodies)
  30. bigqueryio.Write(s, "project", "table", dummyBodies)
  31. }
  32. func main() {
  33. flag.Parse()
  34. beam.Init()
  35. p, s := beam.NewPipelineWithRoot()
  36. buildPipeline(s)
  37. ctx := context.Background()
  38. if err := beamx.Run(ctx, p); err != nil {
  39. log.Exitf(ctx, "Failed to execute pipeline: %v", err)
  40. }
  41. }

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.

huangapple
  • 本文由 发表于 2022年11月1日 19:39:46
  • 转载请务必保留本文链接：https://go.coder-hub.com/74275342.html
  • go
  • google-bigquery
  • google-cloud-dataflow
  • google-cloud-pubsub
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定