实例详解在Go中构建流数据pipeline

package main
 
import (
    "context"
    "fmt"
    "io"
    "os"
 
    "github.com/apache/beam/sdks/go/pkg/beam"
    "github.com/apache/beam/sdks/go/pkg/beam/io/textio"
    "github.com/apache/beam/sdks/go/pkg/beam/runners/direct"
)
 
func main() {
    // 创建Beam管道
    p := beam.NewPipeline()
    s := beam.Impulse(p) // 创建一个简单的触发事件
 
    // 定义管道中的数据源和数据目的地
    lines := textio.Read(s, "./test.txt") // 从文件中读取行
    filtered := beam.ParDo(p, func(line string, emit func(string)) {
        if len(line) > 0 {
            emit(line)
        }
    }, lines) // 应用过滤条件，移除空行
    textio.Write(p, "./output.txt", filtered) // 将处理后的数据写入文件
 
    // 执行管道
    ctx := context.Background()
    if err := direct.Execute(ctx, p); err != nil {
        fmt.Fprintf(os.Stderr, "Failed to execute pipeline: %v", err)
        os.Exit(1)
    }
}

这段代码展示了如何在Go中使用Apache Beam库创建并执行一个简单的数据处理pipeline。它从一个文本文件中读取数据，移除空行，然后将处理后的数据写入到另一个文件中。这个例子简单易懂，并且教会了如何组织和执行数据处理任务。