Go의 Apache Beam ParDo 필터-Golang-php.cn

Go의 Apache Beam ParDo 필터

王林

풀어 주다： 2024-02-05 11:57:58

앞으로

995명이 탐색했습니다.

Go 中的 Apache Beam ParDo 过滤器

질문 내용

저는 Python 개발자이지만 데이터 흐름 파이프라인을 만들려면 go를 사용해야 합니다. Python이나 Java와 비교하여 go를 사용하는 많은 Apache Beam 예제를 찾을 수 없습니다.

사용자 이름과 나이의 구조를 가진 다음 코드가 있습니다. 작업은 연령을 추가한 다음 연령을 기준으로 필터링하는 것입니다. 나이를 늘리는 방법을 찾았으나 필터링 부분에서 막히네요.

으아아아

아래와 같은 함수를 만들어 보았는데, 사용자 객체가 아닌 불리언 값을 반환하게 됩니다. 나는 간단한 것을 놓치고 있다는 것을 알고 있지만 그것을 알아낼 수는 없습니다.

으아아아

파이썬에서는 아래와 같은 함수를 작성할 수 있습니다.

package main

import (
    "context"
    "flag"
    "fmt"

    "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"
)

func init() {
    beam.registerfunction(incrementage)
}

type user struct {
    name string
    age  int
}

func printrow(ctx context.context, list user) {
    fmt.println(list)
}

func incrementage(list user) user {
    list.age++
    return list
}

func main() {

    flag.parse()
    beam.init()

    ctx := context.background()

    p := beam.newpipeline()
    s := p.root()

    var userlist = []user{
        {"bob", 40},
        {"adam", 50},
        {"john", 35},
        {"ben", 8},
    }
    initial := beam.createlist(s, userlist)

    pc := beam.pardo(s, incrementage, initial)

    pc1 := beam.pardo(s, func(row user, emit func(user)) {
        emit(row)
    }, pc)

    beam.pardo0(s, printrow, pc1)

    if err := beamx.run(ctx, p); err != nil {
        log.exitf(ctx, "failed to execute job: %v", err)
    }

}

로그인 후 복사