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/12/18 17:39:38 UTC
(beam) branch master updated: make DataSample external (#29788)
This is an automated email from the ASF dual-hosted git repository.
lostluck 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 7db8c274dd5 make DataSample external (#29788)
7db8c274dd5 is described below
commit 7db8c274dd50642a76dd2bf53ae34aad73127193
Author: Zechen Jiang <ze...@gmail.com>
AuthorDate: Mon Dec 18 09:39:31 2023 -0800
make DataSample external (#29788)
---
sdks/go/pkg/beam/core/runtime/exec/datasampler.go | 28 +++++++++++-----------
.../pkg/beam/core/runtime/exec/datasampler_test.go | 24 +++++++++----------
2 files changed, 26 insertions(+), 26 deletions(-)
diff --git a/sdks/go/pkg/beam/core/runtime/exec/datasampler.go b/sdks/go/pkg/beam/core/runtime/exec/datasampler.go
index 426213a12af..15f32870b49 100644
--- a/sdks/go/pkg/beam/core/runtime/exec/datasampler.go
+++ b/sdks/go/pkg/beam/core/runtime/exec/datasampler.go
@@ -22,7 +22,7 @@ import (
)
// DataSample contains property for sampled element
-type dataSample struct {
+type DataSample struct {
PCollectionID string
Timestamp time.Time
Element []byte
@@ -30,7 +30,7 @@ type dataSample struct {
// DataSampler manages sampled elements based on PCollectionID
type DataSampler struct {
- sampleChannel chan *dataSample
+ sampleChannel chan *DataSample
samplesMap sync.Map // Key: PCollectionID string, Value: *OutputSamples pointer
ctx context.Context
}
@@ -38,7 +38,7 @@ type DataSampler struct {
// NewDataSampler inits a new Data Sampler object and returns pointer to it.
func NewDataSampler(ctx context.Context) *DataSampler {
return &DataSampler{
- sampleChannel: make(chan *dataSample, 1000),
+ sampleChannel: make(chan *DataSample, 1000),
ctx: ctx,
}
}
@@ -57,7 +57,7 @@ func (d *DataSampler) Process() {
// GetSamples returns samples for given pCollectionID.
// If no pCollectionID is provided, return all available samples
-func (d *DataSampler) GetSamples(pids []string) map[string][]*dataSample {
+func (d *DataSampler) GetSamples(pids []string) map[string][]*DataSample {
if len(pids) == 0 {
return d.getAllSamples()
}
@@ -66,7 +66,7 @@ func (d *DataSampler) GetSamples(pids []string) map[string][]*dataSample {
// SendSample is called by PCollection Node to send sampled element to Data Sampler async
func (d *DataSampler) SendSample(pCollectionID string, element []byte, timestamp time.Time) {
- sample := dataSample{
+ sample := DataSample{
PCollectionID: pCollectionID,
Element: element,
Timestamp: timestamp,
@@ -74,8 +74,8 @@ func (d *DataSampler) SendSample(pCollectionID string, element []byte, timestamp
d.sampleChannel <- &sample
}
-func (d *DataSampler) getAllSamples() map[string][]*dataSample {
- var res = make(map[string][]*dataSample)
+func (d *DataSampler) getAllSamples() map[string][]*DataSample {
+ var res = make(map[string][]*DataSample)
d.samplesMap.Range(func(key any, value any) bool {
pid := key.(string)
samples := d.getSamples(pid)
@@ -87,8 +87,8 @@ func (d *DataSampler) getAllSamples() map[string][]*dataSample {
return res
}
-func (d *DataSampler) getSamplesForPCollections(pids []string) map[string][]*dataSample {
- var res = make(map[string][]*dataSample)
+func (d *DataSampler) getSamplesForPCollections(pids []string) map[string][]*DataSample {
+ var res = make(map[string][]*DataSample)
for _, pid := range pids {
samples := d.getSamples(pid)
if len(samples) > 0 {
@@ -98,7 +98,7 @@ func (d *DataSampler) getSamplesForPCollections(pids []string) map[string][]*dat
return res
}
-func (d *DataSampler) addSample(sample *dataSample) {
+func (d *DataSampler) addSample(sample *DataSample) {
p, ok := d.samplesMap.Load(sample.PCollectionID)
if !ok {
p = &outputSamples{maxElements: 10, sampleIndex: 0}
@@ -108,7 +108,7 @@ func (d *DataSampler) addSample(sample *dataSample) {
outputSamples.addSample(sample)
}
-func (d *DataSampler) getSamples(pCollectionID string) []*dataSample {
+func (d *DataSampler) getSamples(pCollectionID string) []*DataSample {
p, ok := d.samplesMap.Load(pCollectionID)
if !ok {
return nil
@@ -118,13 +118,13 @@ func (d *DataSampler) getSamples(pCollectionID string) []*dataSample {
}
type outputSamples struct {
- elements []*dataSample
+ elements []*DataSample
mu sync.Mutex
maxElements int
sampleIndex int
}
-func (o *outputSamples) addSample(element *dataSample) {
+func (o *outputSamples) addSample(element *DataSample) {
o.mu.Lock()
defer o.mu.Unlock()
@@ -136,7 +136,7 @@ func (o *outputSamples) addSample(element *dataSample) {
}
}
-func (o *outputSamples) getSamples() []*dataSample {
+func (o *outputSamples) getSamples() []*DataSample {
o.mu.Lock()
defer o.mu.Unlock()
if len(o.elements) == 0 {
diff --git a/sdks/go/pkg/beam/core/runtime/exec/datasampler_test.go b/sdks/go/pkg/beam/core/runtime/exec/datasampler_test.go
index d648fd89efa..7dce38360af 100644
--- a/sdks/go/pkg/beam/core/runtime/exec/datasampler_test.go
+++ b/sdks/go/pkg/beam/core/runtime/exec/datasampler_test.go
@@ -28,45 +28,45 @@ func TestDataSampler(t *testing.T) {
timestamp := time.Now()
tests := []struct {
name string
- samples []dataSample
+ samples []DataSample
pids []string
- want map[string][]*dataSample
+ want map[string][]*DataSample
}{
{
name: "GetAllSamples",
- samples: []dataSample{
+ samples: []DataSample{
{PCollectionID: "pid1", Element: []byte("element1"), Timestamp: timestamp},
{PCollectionID: "pid2", Element: []byte("element2"), Timestamp: timestamp},
},
pids: []string{},
- want: map[string][]*dataSample{
+ want: map[string][]*DataSample{
"pid1": {{PCollectionID: "pid1", Element: []byte("element1"), Timestamp: timestamp}},
"pid2": {{PCollectionID: "pid2", Element: []byte("element2"), Timestamp: timestamp}},
},
},
{
name: "GetSamplesForPCollections",
- samples: []dataSample{
+ samples: []DataSample{
{PCollectionID: "pid1", Element: []byte("element1"), Timestamp: timestamp},
{PCollectionID: "pid2", Element: []byte("element2"), Timestamp: timestamp},
},
pids: []string{"pid1"},
- want: map[string][]*dataSample{
+ want: map[string][]*DataSample{
"pid1": {{PCollectionID: "pid1", Element: []byte("element1"), Timestamp: timestamp}},
},
},
{
name: "GetSamplesForPCollectionsWithNoResult",
- samples: []dataSample{
+ samples: []DataSample{
{PCollectionID: "pid1", Element: []byte("element1"), Timestamp: timestamp},
{PCollectionID: "pid2", Element: []byte("element2"), Timestamp: timestamp},
},
pids: []string{"pid3"},
- want: map[string][]*dataSample{},
+ want: map[string][]*DataSample{},
},
{
name: "GetSamplesForPCollectionsTooManySamples",
- samples: []dataSample{
+ samples: []DataSample{
{PCollectionID: "pid1", Element: []byte("element1"), Timestamp: timestamp},
{PCollectionID: "pid1", Element: []byte("element2"), Timestamp: timestamp},
{PCollectionID: "pid1", Element: []byte("element3"), Timestamp: timestamp},
@@ -80,7 +80,7 @@ func TestDataSampler(t *testing.T) {
{PCollectionID: "pid1", Element: []byte("element11"), Timestamp: timestamp},
},
pids: []string{"pid1"},
- want: map[string][]*dataSample{
+ want: map[string][]*DataSample{
"pid1": {
{PCollectionID: "pid1", Element: []byte("element2"), Timestamp: timestamp},
{PCollectionID: "pid1", Element: []byte("element3"), Timestamp: timestamp},
@@ -105,7 +105,7 @@ func TestDataSampler(t *testing.T) {
dataSampler.SendSample(sample.PCollectionID, sample.Element, sample.Timestamp)
}
var samplesCount = -1
- var samples map[string][]*dataSample
+ var samples map[string][]*DataSample
for i := 0; i < 5; i++ {
samples = dataSampler.GetSamples(test.pids)
if len(samples) == len(test.want) {
@@ -125,7 +125,7 @@ func TestDataSampler(t *testing.T) {
}
}
-func verifySampledElements(samples, want map[string][]*dataSample) bool {
+func verifySampledElements(samples, want map[string][]*DataSample) bool {
if len(samples) != len(want) {
return false
}