You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by al...@apache.org on 2019/04/04 17:17:41 UTC
[beam] branch master updated: [BEAM-3545] Support Metrics in Go SDK
CombineFns
This is an automated email from the ASF dual-hosted git repository.
altay pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 7014f63 [BEAM-3545] Support Metrics in Go SDK CombineFns
new 0eb9323 Merge pull request #8218 from lostluck/combinermetrics
7014f63 is described below
commit 7014f631c6256c644fcae6ce8adc90a2c1de562b
Author: Robert Burke <ro...@frantil.com>
AuthorDate: Wed Apr 3 22:11:08 2019 +0000
[BEAM-3545] Support Metrics in Go SDK CombineFns
---
sdks/go/pkg/beam/core/metrics/metrics.go | 15 +++++++--
sdks/go/pkg/beam/core/metrics/metrics_test.go | 37 ++++++++++++++++++++++
sdks/go/pkg/beam/core/runtime/exec/combine.go | 41 +++++++++++++++----------
sdks/go/pkg/beam/core/runtime/exec/translate.go | 15 ++++++---
sdks/go/pkg/beam/runners/direct/direct.go | 17 ++++++++--
5 files changed, 99 insertions(+), 26 deletions(-)
diff --git a/sdks/go/pkg/beam/core/metrics/metrics.go b/sdks/go/pkg/beam/core/metrics/metrics.go
index d00f1f0..f80a8e4 100644
--- a/sdks/go/pkg/beam/core/metrics/metrics.go
+++ b/sdks/go/pkg/beam/core/metrics/metrics.go
@@ -82,15 +82,19 @@ func (ctx *beamCtx) Value(key interface{}) interface{} {
switch key {
case bundleKey:
if ctx.bundleID == "" {
- if id := ctx.Value(key); id != nil {
+ if id := ctx.Context.Value(key); id != nil {
ctx.bundleID = id.(string)
+ } else {
+ return nil
}
}
return ctx.bundleID
case ptransformKey:
if ctx.ptransformID == "" {
- if id := ctx.Value(key); id != nil {
+ if id := ctx.Context.Value(key); id != nil {
ctx.ptransformID = id.(string)
+ } else {
+ return nil
}
}
return ctx.ptransformID
@@ -118,8 +122,13 @@ func SetPTransformID(ctx context.Context, id string) context.Context {
return &beamCtx{Context: ctx, ptransformID: id}
}
+const (
+ bundleIdUnset = "(bundle id unset)"
+ ptransformIdUnset = "(ptransform id unset)"
+)
+
func getContextKey(ctx context.Context, n name) key {
- key := key{name: n, bundle: "(bundle id unset)", ptransform: "(ptransform id unset)"}
+ key := key{name: n, bundle: bundleIdUnset, ptransform: ptransformIdUnset}
if id := ctx.Value(bundleKey); id != nil {
key.bundle = id.(string)
}
diff --git a/sdks/go/pkg/beam/core/metrics/metrics_test.go b/sdks/go/pkg/beam/core/metrics/metrics_test.go
index 6e28981..ec10720 100644
--- a/sdks/go/pkg/beam/core/metrics/metrics_test.go
+++ b/sdks/go/pkg/beam/core/metrics/metrics_test.go
@@ -25,6 +25,43 @@ import (
// bID is a bundleId to use in the tests, if nothing more specific is needed.
const bID = "bID"
+// TestRobustness validates metrics not panicking if the context doesn't
+// have the bundle or transform ID.
+func TestRobustness(t *testing.T) {
+ m := NewCounter("Test", "myCount")
+ m.Inc(context.Background(), 3)
+ ptCtx := SetPTransformID(context.Background(), "MY_TRANSFORM")
+ m.Inc(ptCtx, 3)
+ bCtx := SetBundleID(context.Background(), bID)
+ m.Inc(bCtx, 3)
+}
+
+func TestBeamContext(t *testing.T) {
+ t.Run("ptransformID", func(t *testing.T) {
+ ptID := "MY_TRANSFORM"
+ ctx := SetPTransformID(context.Background(), ptID)
+ key := getContextKey(ctx, name{})
+ if key.bundle != bundleIdUnset {
+ t.Errorf("bundleId = %q, want %q", key.bundle, bundleIdUnset)
+ }
+ if key.ptransform != ptID {
+ t.Errorf("ptransformId = %q, want %q", key.ptransform, ptID)
+ }
+
+ })
+
+ t.Run("bundleID", func(t *testing.T) {
+ ctx := SetBundleID(context.Background(), bID)
+ key := getContextKey(ctx, name{})
+ if key.bundle != bID {
+ t.Errorf("bundleId = %q, want %q", key.bundle, bID)
+ }
+ if key.ptransform != ptransformIdUnset {
+ t.Errorf("ptransformId = %q, want %q", key.ptransform, ptransformIdUnset)
+ }
+ })
+}
+
func ctxWith(b, pt string) context.Context {
ctx := context.Background()
ctx = SetPTransformID(ctx, pt)
diff --git a/sdks/go/pkg/beam/core/runtime/exec/combine.go b/sdks/go/pkg/beam/core/runtime/exec/combine.go
index ce7b33d..505b531 100644
--- a/sdks/go/pkg/beam/core/runtime/exec/combine.go
+++ b/sdks/go/pkg/beam/core/runtime/exec/combine.go
@@ -24,6 +24,7 @@ import (
"github.com/apache/beam/sdks/go/pkg/beam/core/graph"
"github.com/apache/beam/sdks/go/pkg/beam/core/graph/coder"
+ "github.com/apache/beam/sdks/go/pkg/beam/core/metrics"
"github.com/apache/beam/sdks/go/pkg/beam/core/typex"
"github.com/apache/beam/sdks/go/pkg/beam/core/util/reflectx"
"github.com/apache/beam/sdks/go/pkg/beam/util/errorx"
@@ -36,6 +37,9 @@ type Combine struct {
UsesKey bool
Out Node
+ PID string
+ ctx context.Context
+
binaryMergeFn reflectx.Func2x1 // optimized caller in the case of binary merge accumulators
status Status
@@ -106,7 +110,12 @@ func (n *Combine) StartBundle(ctx context.Context, id string, data DataContext)
}
n.status = Active
- if err := n.Out.StartBundle(ctx, id, data); err != nil {
+ // Allocating contexts all the time is expensive, but we seldom re-write them,
+ // and never accept modified contexts from users, so we will cache them per-bundle
+ // per-unit, to avoid the constant allocation overhead.
+ n.ctx = metrics.SetPTransformID(ctx, n.PID)
+
+ if err := n.Out.StartBundle(n.ctx, id, data); err != nil {
return n.fail(err)
}
return nil
@@ -122,7 +131,7 @@ func (n *Combine) ProcessElement(ctx context.Context, value *FullValue, values .
// Note that we do not explicitly call merge, although it may
// be called implicitly when adding input.
- a, err := n.newAccum(ctx, value.Elm)
+ a, err := n.newAccum(n.ctx, value.Elm)
if err != nil {
return n.fail(err)
}
@@ -142,18 +151,18 @@ func (n *Combine) ProcessElement(ctx context.Context, value *FullValue, values .
return n.fail(err)
}
- a, err = n.addInput(ctx, a, value.Elm, v.Elm, value.Timestamp, first)
+ a, err = n.addInput(n.ctx, a, value.Elm, v.Elm, value.Timestamp, first)
if err != nil {
return n.fail(err)
}
first = false
}
- out, err := n.extract(ctx, a)
+ out, err := n.extract(n.ctx, a)
if err != nil {
return n.fail(err)
}
- return n.Out.ProcessElement(ctx, &FullValue{Windows: value.Windows, Elm: value.Elm, Elm2: out, Timestamp: value.Timestamp})
+ return n.Out.ProcessElement(n.ctx, &FullValue{Windows: value.Windows, Elm: value.Elm, Elm2: out, Timestamp: value.Timestamp})
}
// FinishBundle completes this node's processing of a bundle.
@@ -175,7 +184,7 @@ func (n *Combine) FinishBundle(ctx context.Context) error {
n.extractOutputInv.Reset()
}
- if err := n.Out.FinishBundle(ctx); err != nil {
+ if err := n.Out.FinishBundle(n.ctx); err != nil {
return n.fail(err)
}
return nil
@@ -336,14 +345,14 @@ func (n *LiftedCombine) ProcessElement(ctx context.Context, value *FullValue, va
if notfirst {
a = afv.Elm2
} else {
- b, err := n.newAccum(ctx, value.Elm)
+ b, err := n.newAccum(n.Combine.ctx, value.Elm)
if err != nil {
return n.fail(err)
}
a = b
}
- a, err = n.addInput(ctx, a, value.Elm, value.Elm2, value.Timestamp, !notfirst)
+ a, err = n.addInput(n.Combine.ctx, a, value.Elm, value.Elm2, value.Timestamp, !notfirst)
if err != nil {
return n.fail(err)
}
@@ -363,7 +372,7 @@ func (n *LiftedCombine) ProcessElement(ctx context.Context, value *FullValue, va
if k == key {
continue
}
- if err := n.Out.ProcessElement(ctx, &a); err != nil {
+ if err := n.Out.ProcessElement(n.Combine.ctx, &a); err != nil {
return err
}
delete(n.cache, k)
@@ -388,7 +397,7 @@ func (n *LiftedCombine) FinishBundle(ctx context.Context) error {
// Need to run n.Out.ProcessElement for all the cached precombined KVs, and
// then finally Finish bundle as normal.
for _, a := range n.cache {
- if err := n.Out.ProcessElement(ctx, &a); err != nil {
+ if err := n.Out.ProcessElement(n.Combine.ctx, &a); err != nil {
return err
}
}
@@ -396,7 +405,7 @@ func (n *LiftedCombine) FinishBundle(ctx context.Context) error {
// Down isn't guaranteed to be called.
n.cache = nil
- return n.Combine.FinishBundle(ctx)
+ return n.Combine.FinishBundle(n.Combine.ctx)
}
// Down tears down the cache.
@@ -423,7 +432,7 @@ func (n *MergeAccumulators) ProcessElement(ctx context.Context, value *FullValue
if n.status != Active {
return fmt.Errorf("invalid status for combine merge %v: %v", n.UID, n.status)
}
- a, err := n.newAccum(ctx, value.Elm)
+ a, err := n.newAccum(n.Combine.ctx, value.Elm)
if err != nil {
return n.fail(err)
}
@@ -447,12 +456,12 @@ func (n *MergeAccumulators) ProcessElement(ctx context.Context, value *FullValue
first = false
continue
}
- a, err = n.mergeAccumulators(ctx, a, v.Elm)
+ a, err = n.mergeAccumulators(n.Combine.ctx, a, v.Elm)
if err != nil {
return err
}
}
- return n.Out.ProcessElement(ctx, &FullValue{Windows: value.Windows, Elm: value.Elm, Elm2: a, Timestamp: value.Timestamp})
+ return n.Out.ProcessElement(n.Combine.ctx, &FullValue{Windows: value.Windows, Elm: value.Elm, Elm2: a, Timestamp: value.Timestamp})
}
// Up eagerly gets the optimized binary merge function.
@@ -478,9 +487,9 @@ func (n *ExtractOutput) ProcessElement(ctx context.Context, value *FullValue, va
if n.status != Active {
return fmt.Errorf("invalid status for combine extract %v: %v", n.UID, n.status)
}
- out, err := n.extract(ctx, value.Elm2)
+ out, err := n.extract(n.Combine.ctx, value.Elm2)
if err != nil {
return n.fail(err)
}
- return n.Out.ProcessElement(ctx, &FullValue{Windows: value.Windows, Elm: value.Elm, Elm2: out, Timestamp: value.Timestamp})
+ return n.Out.ProcessElement(n.Combine.ctx, &FullValue{Windows: value.Windows, Elm: value.Elm, Elm2: out, Timestamp: value.Timestamp})
}
diff --git a/sdks/go/pkg/beam/core/runtime/exec/translate.go b/sdks/go/pkg/beam/core/runtime/exec/translate.go
index 031ac82..955ff5b 100644
--- a/sdks/go/pkg/beam/core/runtime/exec/translate.go
+++ b/sdks/go/pkg/beam/core/runtime/exec/translate.go
@@ -375,13 +375,15 @@ func (b *builder) makeLink(from string, id linkID) (Node, error) {
switch op {
case graph.ParDo:
- n := &ParDo{UID: b.idgen.New(), PID: id.to, Inbound: in, Out: out}
+ n := &ParDo{UID: b.idgen.New(), PID: transform.UniqueName, Inbound: in, Out: out}
n.Fn, err = graph.AsDoFn(fn)
if err != nil {
return nil, err
}
- // TODO(lostluck): 2018/03/22 Look into why transform.UniqueName isn't populated at this point, and switch n.PID to that instead.
- n.PID = path.Base(n.Fn.Name())
+ if n.PID == "" {
+ // Handle the case where transform.UniqueName isn't populated.
+ n.PID = path.Base(n.Fn.Name())
+ }
input := unmarshalKeyedValues(transform.GetInputs())
for i := 1; i < len(input); i++ {
@@ -406,13 +408,18 @@ func (b *builder) makeLink(from string, id linkID) (Node, error) {
u = n
case graph.Combine:
- cn := &Combine{UID: b.idgen.New(), Out: out[0]}
+ cn := &Combine{UID: b.idgen.New(), PID: transform.UniqueName, Out: out[0]}
cn.Fn, err = graph.AsCombineFn(fn)
if err != nil {
return nil, err
}
cn.UsesKey = typex.IsKV(in[0].Type)
+ if cn.PID == "" {
+ // Handle the case where transform.UniqueName isn't populated.
+ cn.PID = path.Base(cn.Fn.Name())
+ }
+
switch urn {
case urnPerKeyCombinePre:
inputs := unmarshalKeyedValues(transform.GetInputs())
diff --git a/sdks/go/pkg/beam/runners/direct/direct.go b/sdks/go/pkg/beam/runners/direct/direct.go
index f456391..608fe98 100644
--- a/sdks/go/pkg/beam/runners/direct/direct.go
+++ b/sdks/go/pkg/beam/runners/direct/direct.go
@@ -220,8 +220,13 @@ func (b *builder) makeLink(id linkID) (exec.Node, error) {
var u exec.Node
switch edge.Op {
case graph.ParDo:
- pardo := &exec.ParDo{UID: b.idgen.New(), Fn: edge.DoFn, Inbound: edge.Input, Out: out}
- pardo.PID = path.Base(pardo.Fn.Name())
+ pardo := &exec.ParDo{
+ UID: b.idgen.New(),
+ Fn: edge.DoFn,
+ Inbound: edge.Input,
+ Out: out,
+ PID: path.Base(edge.DoFn.Name()),
+ }
if len(edge.Input) == 1 {
u = pardo
break
@@ -249,7 +254,13 @@ func (b *builder) makeLink(id linkID) (exec.Node, error) {
case graph.Combine:
usesKey := typex.IsKV(edge.Input[0].Type)
- u = &exec.Combine{UID: b.idgen.New(), Fn: edge.CombineFn, UsesKey: usesKey, Out: out[0]}
+ u = &exec.Combine{
+ UID: b.idgen.New(),
+ Fn: edge.CombineFn,
+ UsesKey: usesKey,
+ Out: out[0],
+ PID: path.Base(edge.CombineFn.Name()),
+ }
case graph.CoGBK:
u = &CoGBK{UID: b.idgen.New(), Edge: edge, Out: out[0]}