我正在嘗試從 PubSub 讀取訊息，然后寫入 DataFlow 中的 BigQuery 表。但是，我使用直接運行器遇到了“無根單元”錯誤。
這是我的代碼；
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 執行管道失敗：翻譯失敗原因：沒有根單元退出狀態 1
uj5u.com熱心網友回復：
pubsubio 的當前實作僅適用于 Dataflow Runner。
轉載請註明出處，本文鏈接：https://www.uj5u.com/caozuo/526308.html
標籤：Google Cloud Collective 去谷歌大查询谷歌云数据流谷歌云发布订阅