Go语言中Apache Beam读取BQ并发布到PubSub报错求助

问题根源 #

报错提示Could not find the sink for pubsub, Check that the sink library specifies alwayslink = 1.，本质是Go编译器的代码优化特性：如果代码中没有显式直接引用某个包的符号，编译器会自动移除该包的代码。而Apache Beam的PubSub连接器属于外部扩展组件，Dataflow运行时需要这些代码，但默认编译时被优化掉了。

具体解决步骤 #

1. 强制链接PubSub连接器 #

在代码的导入部分，添加下划线导入确保PubSub包的初始化代码被执行，避免被编译器优化：

import (
    // ... 其他导入
    _ "github.com/apache/beam/sdks/v2/go/pkg/beam/io/pubsubio"
)

注：即使你已经导入了pubsubio，下划线导入可以强制触发包的init逻辑，确保连接器注册到Beam中

2. 使用Dataflow编译标签运行作业 #

提交作业时，添加-tags=dataflow标签，确保Dataflow相关的连接器代码被正确编译：

go run -tags=dataflow main.go --project=PROJECTID --runner=dataflow --region=us-east1 --staging_location=gs://PROJECTID/tmp

3. 简化冗余代码 #

你的代码中有一个无意义的ParDo操作，只是原样转发数据，完全可以删除，直接将BigQuery查询结果传给转换函数：

// 移除多余的ParDo
// pc := beam.ParDo(s, func(row CommentRow, emit func(CommentRow)) {
//     emit(row)
// }, rows)

// 直接使用rows作为输入
pc1 := beam.ParDo(s, typeConvert, rows)

4. 修正PubSub Topic路径（可选） #

避免硬编码项目ID，用变量动态拼接Topic路径，减少人为错误：

import "fmt"

// ...

topic := fmt.Sprintf("projects/%s/topics/test-topic", project)
pubsubio.Write(s, project, topic, pc1)

验证修改后的完整代码 #

package main

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

    "github.com/apache/beam/sdks/v2/go/pkg/beam"
    "github.com/apache/beam/sdks/v2/go/pkg/beam/io/bigqueryio"
    "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/options/gcpopts"
    "github.com/apache/beam/sdks/v2/go/pkg/beam/x/beamx"
    _ "github.com/apache/beam/sdks/v2/go/pkg/beam/io/pubsubio"
)

// CommentRow models 1 row of HackerNews comments.
type CommentRow struct {
    Text string `bigquery:"text"`
}

func init() {
    beam.RegisterFunction(typeConvert)
}

const query = `SELECT text
FROM ` + "`bigquery-public-data.hacker_news.comments`" + `
WHERE time_ts BETWEEN '2013-01-01' AND '2014-01-01' and text IS NOT NULL
LIMIT 1000
`

func typeConvert(list CommentRow) []byte {
    b, _ := json.Marshal(list)
    return b
}

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

    ctx := context.Background()
    p := beam.NewPipeline()
    s := p.Root()
    project := gcpopts.GetProject(ctx)

    // Build a PCollection<CommentRow> by querying BigQuery.
    rows := bigqueryio.Query(s, project, query,
        reflect.TypeOf(CommentRow{}), bigqueryio.UseStandardSQL())

    pc1 := beam.ParDo(s, typeConvert, rows)

    topic := fmt.Sprintf("projects/%s/topics/test-topic", project)
    pubsubio.Write(s, project, topic, pc1)

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

内容的提问来源于stack exchange，提问作者Ashok KS