Apache Beam meninggalkan sertai dalam Go

editor php Xiaoxin berada di sini untuk memperkenalkan kepada anda Apache Beam yang tinggal serta dalam bahasa Go. Apache Beam ialah rangka kerja pemprosesan data teragih yang menyediakan model pengaturcaraan biasa untuk melaksanakan tugas pemprosesan kelompok dan strim pada enjin pemprosesan data teragih yang berbeza. Cantuman kiri ialah operasi pemprosesan data biasa Ia boleh mengaitkan dua set data mengikut kunci tertentu dan mengembalikan semua rekod dalam set data kiri dan rekod sepadan dalam set data yang betul. Artikel ini akan memperkenalkan secara terperinci cara menggunakan Apache Beam untuk melaksanakan operasi gabungan kiri dalam bahasa Go.

Kandungan soalan

Adakah terdapat cara mudah untuk melakukan gabungan kiri 2 koleksi menggunakan go? Saya mendapati bahawa sambungan sql hanya tersedia di java.

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
}

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)

    // Left Join custPcol with orderPCol
    // Expected Result
    // CustID | FName   |OrderID| Amount
    //     1  | Bob     |   123 | 100
    //     2  | Adam    |       |
    //     3  | John    |   125 | 100
    //     4  | Ben     |       |
    //     5  | Jose    |       |
    //     6  | Bryan   |       |
    //     7  | Kim     |   125 | 100
    //     8  | Tim     |       |

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

}

Saya ingin menyertai 2 koleksi ini dan melakukan operasi selanjutnya. Saya melihat dokumentasi untuk cogroupbykey tetapi saya tidak boleh menukarnya kepada format yang boleh dilakukan oleh sql join biasa.

Ada cadangan tentang perkara ini?

Penyelesaian

Cuba ini

type resulttype struct {
    custid  int
    fname   string
    orderid int
    amount  int
}

result := beam.pardo(s, func(c customer, iterorder func(*order) bool) resulttype {
    var o order

    for iterorder(&o) {
        if c.custid == o.cust_id {
            return resulttype{
                custid:  c.custid,
                fname:   c.fname,
                orderid: o.orderid,
                amount:  o.amount,
            }
        }
    }

    return resulttype{
        custid: c.custid,
        fname:  c.fname,
    }
}, custpcol, beam.sideinput{input: orderpcol})

Atau kalau nak guna cogroupbykey...

custWithKeyPCol := beam.ParDo(s, func(c customer) (int, customer) {
    return c.CustID, c
}, custPCol)

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

resultPCol := beam.CoGroupByKey(s, custWithKeyPCol, orderWithKeyPCol)

beam.ParDo0(s, func(CustID int, custIter func(*customer) bool, orderIter func(*order) bool) {
    c, o := customer{}, order{}
    for custIter(&c) {
        if ok := orderIter(&o); ok {
            fmt.Println(CustID, c.FName, o.OrderID, o.Amount)
        }
        fmt.Println(CustID, c.FName)
    }
}, resultPCol)

Atas ialah kandungan terperinci Apache Beam meninggalkan sertai dalam Go. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!