You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by lo...@apache.org on 2023/10/28 03:37:19 UTC
(beam) 01/01: minor prism fixes
This is an automated email from the ASF dual-hosted git repository.
lostluck pushed a commit to branch minorPrismFixes
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 809517d98894c9f14a9d7be2255bd457725533ad
Author: Robert Burke <ro...@frantil.com>
AuthorDate: Wed Oct 25 06:51:35 2023 -0700
minor prism fixes
---
.../runners/prism/internal/jobservices/metrics.go | 2 +-
sdks/go/pkg/beam/runners/prism/internal/web/web.go | 9 +++++++
.../beam/runners/prism/internal/worker/bundle.go | 30 +++++++++++-----------
3 files changed, 25 insertions(+), 16 deletions(-)
diff --git a/sdks/go/pkg/beam/runners/prism/internal/jobservices/metrics.go b/sdks/go/pkg/beam/runners/prism/internal/jobservices/metrics.go
index 6db16191de9..e0caec55881 100644
--- a/sdks/go/pkg/beam/runners/prism/internal/jobservices/metrics.go
+++ b/sdks/go/pkg/beam/runners/prism/internal/jobservices/metrics.go
@@ -475,8 +475,8 @@ func (m *metricsStore) AddShortIDs(resp *fnpb.MonitoringInfosMetadataResponse) {
m.shortIDsToKeys[short] = key
}
for d, payloads := range m.unprocessedPayloads {
- m.contributeMetrics(durability(d), payloads)
m.unprocessedPayloads[d] = nil
+ m.contributeMetrics(durability(d), payloads)
}
}
diff --git a/sdks/go/pkg/beam/runners/prism/internal/web/web.go b/sdks/go/pkg/beam/runners/prism/internal/web/web.go
index b7afad35aee..765f0b50c83 100644
--- a/sdks/go/pkg/beam/runners/prism/internal/web/web.go
+++ b/sdks/go/pkg/beam/runners/prism/internal/web/web.go
@@ -230,6 +230,9 @@ func (h *jobDetailsHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
var msecMets []string
// TODO: Figure out where uniquename or id is being used in prism. It should be all global transform ID to faciliate lookups.
+ for _, msec := range msecs[id] {
+ msecMets = append(msecMets, fmt.Sprintf("\n- %+v", msec.Result()))
+ }
for _, msec := range msecs[pt.GetUniqueName()] {
msecMets = append(msecMets, fmt.Sprintf("\n- %+v", msec.Result()))
}
@@ -239,6 +242,12 @@ func (h *jobDetailsHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
}
var userMetrics []string
+ for _, ctr := range counters[id] {
+ userMetrics = append(userMetrics, fmt.Sprintf("\n- %s.%s: %v", ctr.Namespace(), ctr.Name(), ctr.Result()))
+ }
+ for _, dist := range distributions[id] {
+ userMetrics = append(userMetrics, fmt.Sprintf("\n- %s.%s: %+v", dist.Namespace(), dist.Name(), dist.Result()))
+ }
for _, ctr := range counters[pt.GetUniqueName()] {
userMetrics = append(userMetrics, fmt.Sprintf("\n- %s.%s: %v", ctr.Namespace(), ctr.Name(), ctr.Result()))
}
diff --git a/sdks/go/pkg/beam/runners/prism/internal/worker/bundle.go b/sdks/go/pkg/beam/runners/prism/internal/worker/bundle.go
index 573bdf4aeb9..fab8cbc141f 100644
--- a/sdks/go/pkg/beam/runners/prism/internal/worker/bundle.go
+++ b/sdks/go/pkg/beam/runners/prism/internal/worker/bundle.go
@@ -16,6 +16,7 @@
package worker
import (
+ "bytes"
"context"
"fmt"
"sync/atomic"
@@ -126,22 +127,21 @@ func (b *B) ProcessOn(ctx context.Context, wk *W) <-chan struct{} {
}
// TODO: make batching decisions.
- for i, d := range b.InputData {
- select {
- case wk.DataReqs <- &fnpb.Elements{
- Data: []*fnpb.Elements_Data{
- {
- InstructionId: b.InstID,
- TransformId: b.InputTransformID,
- Data: d,
- IsLast: i+1 == len(b.InputData),
- },
+ dataBuf := bytes.Join(b.InputData, []byte{})
+ select {
+ case wk.DataReqs <- &fnpb.Elements{
+ Data: []*fnpb.Elements_Data{
+ {
+ InstructionId: b.InstID,
+ TransformId: b.InputTransformID,
+ Data: dataBuf,
+ IsLast: true,
},
- }:
- case <-ctx.Done():
- b.DataDone()
- return b.DataWait
- }
+ },
+ }:
+ case <-ctx.Done():
+ b.DataDone()
+ return b.DataWait
}
return b.DataWait
}