如何在Go语言中实现Apache Beam两个PCollection的左连接？

在Go的Apache Beam中确实没有直接的SQL Join API（目前仅Java SDK支持），但可以通过CoGroupByKey配合后续的转换操作来实现左连接，以下是针对你代码的完整实现方案：

步骤1：定义左连接结果结构体 #

首先定义匹配预期输出格式的结果结构体：

type JoinResult struct {
    CustID  int
    FName   string
    OrderID int // 无对应订单时为0，若需明确区分空值可改为*int类型
    Amount  int // 无对应订单时为0，若需明确区分空值可改为*int类型
}

步骤2：转换PCollection为KV结构 #

将两个原始PCollection转换为以客户ID为键的KV结构，为分组做准备：

// 将客户集合转换为KV<CustID, customer>
custKV := beam.ParDo(s, func(c customer) (int, customer) {
    return c.CustID, c
}, custPCol)

// 将订单集合转换为KV<Cust_ID, order>
orderKV := beam.ParDo(s, func(o order) (int, order) {
    return o.Cust_ID, o
}, orderPCol)

步骤3：使用CoGroupByKey分组 #

通过CoGroupByKey将同一客户ID对应的客户信息和订单信息聚合到一起：

// 按客户ID分组，得到每个ID对应的客户列表和订单列表
grouped := beam.CoGroupByKey(s, custKV, orderKV)

步骤4：展开分组结果生成左连接数据 #

通过ParDo遍历每个分组，生成符合左连接逻辑的结果：

// 处理分组数据，生成左连接结果
joined := beam.ParDo(s, func(key int, values beam.CoGroupedValues[customer, order]) []JoinResult {
    // 假设CustID唯一，取第一个客户信息
    custs := values.Get1()
    if len(custs) == 0 {
        return nil
    }
    cust := custs[0]
    
    orders := values.Get2()
    results := make([]JoinResult, 0)
    
    // 有对应订单时，逐个生成结果行
    if len(orders) > 0 {
        for _, o := range orders {
            results = append(results, JoinResult{
                CustID:  cust.CustID,
                FName:   cust.FName,
                OrderID: o.OrderID,
                Amount:  o.Amount,
            })
        }
    } else {
        // 无对应订单时，生成填充默认值的结果行
        results = append(results, JoinResult{
            CustID: cust.CustID,
            FName:  cust.FName,
        })
    }
    return results
}, grouped)

完整修改后的代码 #

替换原代码中注释部分的内容后，完整代码如下：

package main

import (
    "context"
    "flag"

    "github.com/apache/beam/sdks/v2/go/pkg/beam"
    "github.com/apache/beam/sdks/v2/go/pkg/beam/log"
    "github.com/apache/beam/sdks/v2/go/pkg/beam/x/beamx"
)

type customer struct {
    CustID int
    FName  string
}

type order struct {
    OrderID int
    Amount  int
    Cust_ID int
}

type JoinResult struct {
    CustID  int
    FName   string
    OrderID int
    Amount  int
}

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

    ctx := context.Background()

    p := beam.NewPipeline()
    s := p.Root()

    var custList = []customer{
        {1, "Bob"},
        {2, "Adam"},
        {3, "John"},
        {4, "Ben"},
        {5, "Jose"},
        {6, "Bryan"},
        {7, "Kim"},
        {8, "Tim"},
    }

    var orderList = []order{
        {123, 100, 1},
        {125, 30, 3},
        {128, 50, 7},
    }

    custPCol := beam.CreateList(s, custList)
    orderPCol := beam.CreateList(s, orderList)

    // 转换为KV结构
    custKV := beam.ParDo(s, func(c customer) (int, customer) {
        return c.CustID, c
    }, custPCol)

    orderKV := beam.ParDo(s, func(o order) (int, order) {
        return o.Cust_ID, o
    }, orderPCol)

    // 按客户ID分组
    grouped := beam.CoGroupByKey(s, custKV, orderKV)

    // 生成左连接结果
    joined := beam.ParDo(s, func(key int, values beam.CoGroupedValues[customer, order]) []JoinResult {
        custs := values.Get1()
        if len(custs) == 0 {
            return nil
        }
        cust := custs[0]
        
        orders := values.Get2()
        results := make([]JoinResult, 0)
        
        if len(orders) > 0 {
            for _, o := range orders {
                results = append(results, JoinResult{
                    CustID:  cust.CustID,
                    FName:   cust.FName,
                    OrderID: o.OrderID,
                    Amount:  o.Amount,
                })
            }
        } else {
            results = append(results, JoinResult{
                CustID: cust.CustID,
                FName:  cust.FName,
            })
        }
        return results
    }, grouped)

    // 可选：打印结果验证
    beam.ParDo(s, func(result JoinResult) {
        log.Infof(ctx, "CustID: %d | FName: %s | OrderID: %d | Amount: %d", result.CustID, result.FName, result.OrderID, result.Amount)
    }, joined)

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

关键逻辑说明 #

• KV转换：确保两个集合使用相同的连接键（客户ID），这是CoGroupByKey能够正确分组的前提。
• CoGroupByKey：将同一键对应的两类数据聚合，得到每个客户ID对应的客户信息和所有关联订单。
• 结果展开：遍历每个分组，保证每个客户至少生成一行结果——有订单则匹配生成对应行，无订单则生成填充默认值的行，完全符合SQL左连接的行为。

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