[BEAM-9679] Add Side Input to Core Transform Go SDK katas (apache#12024)

learning/katas/go/Core Transforms/Side Input/Side Input/cmd/main.go

@@ -0,0 +1,57 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+	"context"
+	"github.com/apache/beam/sdks/go/pkg/beam"
+	"github.com/apache/beam/sdks/go/pkg/beam/log"
+	"github.com/apache/beam/sdks/go/pkg/beam/x/beamx"
+	"github.com/apache/beam/sdks/go/pkg/beam/x/debug"
+	"side_input/pkg/task"
+)
+
+func main() {
+	ctx := context.Background()
+
+	p, s := beam.NewPipelineWithRoot()
+
+	citiesToCountriesKV := beam.ParDo(s, func(_ []byte, emit func(string, string)){
+		emit("Beijing", "China")
+		emit("London", "United Kingdom")
+		emit("San Francisco", "United States")
+		emit("Singapore", "Singapore")
+		emit("Sydney", "Australia")
+	}, beam.Impulse(s))
+
+	persons := beam.Create(s,
+		task.Person{Name: "Henry", City: "Singapore"},
+		task.Person{Name: "Jane", City: "San Francisco"},
+		task.Person{Name: "Lee", City: "Beijing"},
+		task.Person{Name: "John", City: "Sydney"},
+		task.Person{Name: "Alfred", City: "London"},
+	)
+
+	output := task.ApplyTransform(s, persons, citiesToCountriesKV)
+
+	debug.Print(s, output)
+
+	err := beamx.Run(ctx, p)
+
+	if err != nil {
+		log.Exitf(context.Background(), "Failed to execute job: %v", err)
+	}
+}

learning/katas/go/Core Transforms/Side Input/Side Input/go.mod

@@ -0,0 +1,31 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+module side_input
+
+go 1.14
+
+require (
+	github.com/apache/beam v2.22.0+incompatible
+	github.com/golang/protobuf v1.4.2 // indirect
+	github.com/google/go-cmp v0.4.1 // indirect
+	github.com/googleapis/gax-go v1.0.3 // indirect
+	golang.org/x/net v0.0.0-20200602114024-627f9648deb9 // indirect
+	golang.org/x/sys v0.0.0-20200615200032-f1bc736245b1 // indirect
+	golang.org/x/text v0.3.3 // indirect
+	google.golang.org/api v0.27.0 // indirect
+	google.golang.org/genproto v0.0.0-20200616192300-fc83d8c00726 // indirect
+	google.golang.org/grpc v1.29.1 // indirect
+)

learning/katas/go/Core Transforms/Side Input/Side Input/pkg/task/task.go

@@ -0,0 +1,46 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package task
+
+import (
+	"github.com/apache/beam/sdks/go/pkg/beam"
+)
+
+type Person struct {
+	Name, City, Country string
+}
+
+func ApplyTransform(s beam.Scope, personsKV beam.PCollection, citiesToCountries beam.PCollection) beam.PCollection {
+	citiesToCountriesView := beam.SideInput{
+		Input: citiesToCountries,
+	}
+	return beam.ParDo(s, joinFn, personsKV, citiesToCountriesView)
+}
+
+func joinFn(person Person, citiesToCountriesIter func(*string, *string) bool, emit func(Person)) {
+	var city, country string
+	for citiesToCountriesIter(&city, &country) {
+		if person.City == city {
+			emit(Person{
+				Name:    person.Name,
+				City:    city,
+				Country: country,
+			})
+			break
+		}
+	}
+}
+

learning/katas/go/Core Transforms/Side Input/Side Input/task-info.yaml

@@ -0,0 +1,41 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#  http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+type: edu
+files:
+- name: test/task_test.go
+  visible: false
+- name: cmd/main.go
+  visible: true
+- name: go.mod
+  visible: false
+- name: go.sum
+  visible: false
+- name: pkg/task/task.go
+  visible: true
+  placeholders:
+  - offset: 1059
+    length: 46
+    placeholder_text: TODO()
+  - offset: 1185
+    length: 83
+    placeholder_text: TODO()
+  - offset: 1273
+    length: 202
+    placeholder_text: TODO()

learning/katas/go/Core Transforms/Side Input/Side Input/task-remote-info.yaml

@@ -0,0 +1,2 @@
+id: 1387150
+update_date: Mon, 22 Jun 2020 02:53:16 UTC

learning/katas/go/Core Transforms/Side Input/Side Input/task.md

@@ -0,0 +1,68 @@
+<!--
+    Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements.  See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership.  The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License.  You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied.  See the License for the
+    specific language governing permissions and limitations
+    under the License.
+-->
+
+# Side Input
+
+In addition to the main input PCollection, you can provide additional inputs to a ParDo transform 
+in the form of side inputs. A side input is an additional input that your DoFn can access each time 
+it processes an element in the input PCollection. When you specify a side input, you create a view 
+of some other data that can be read from within the ParDo transform’s DoFn while processing each 
+element.
+
+Side inputs are useful if your ParDo needs to inject additional data when processing each element 
+in the input PCollection, but the additional data needs to be determined at runtime (and not 
+hard-coded). Such values might be determined by the input data, or depend on a different branch of 
+your pipeline.
+
+**Kata:** Please enrich each Person with the country based on the city he/she lives in.
+
+<div class="hint">
+    Use <a href="https://godoc.org/github.com/apache/beam/sdks/go/pkg/beam#hdr-Side_Inputs">
+    beam.SideInput</a> to create a side input of the citiesToCountries PCollection
+</div>
+
+<div class="hint">
+    Use <a href="https://godoc.org/github.com/apache/beam/sdks/go/pkg/beam#ParDo">
+    beam.ParDo</a> with a DoFn that accepts a
+    <a href="https://godoc.org/github.com/apache/beam/sdks/go/pkg/beam#hdr-Side_Inputs">
+    beam.SideInput</a>.
+</div>
+
+<div class="hint">
+    Refer to the Beam Programming Guide
+    <a href="https://beam.apache.org/documentation/programming-guide/#side-inputs">"Side inputs"</a>
+    section for more information.
+</div>
+
+<div class="hint">
+    A ParDo that has a PCollection KV as side input expects a DoFn that looks like the following.    
+
+```
+func doFn(element X, kvIterator func(*K, *V) bool, emit func(Y)) {
+    // element of type X comes from your PCollection input
+    // from your KV side input:     
+    var key K   // expect a key of type K 
+    var value V // expect a value of type V 
+    for kvIterator(&key, &value) {
+        // do something with key and value
+        // emit(some result)
+    }
+} 
+``` 
+</div>

learning/katas/go/Core Transforms/Side Input/Side Input/test/task_test.go

@@ -0,0 +1,95 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package test
+
+import (
+	"github.com/apache/beam/sdks/go/pkg/beam"
+	"github.com/apache/beam/sdks/go/pkg/beam/testing/passert"
+	"github.com/apache/beam/sdks/go/pkg/beam/testing/ptest"
+	"side_input/pkg/task"
+	"testing"
+)
+
+func TestApplyTransform(t *testing.T) {
+	p, s := beam.NewPipelineWithRoot()
+
+	tests := []struct {
+		args struct {
+			citiesAndCountriesKV beam.PCollection
+			persons              beam.PCollection
+		}
+		want  []interface{}
+	}{
+		{
+			args: struct {
+				citiesAndCountriesKV beam.PCollection
+				persons            beam.PCollection
+			}{
+				citiesAndCountriesKV: beam.ParDo(
+					s.Scope("Cities and Countries"),
+					func(_ []byte, emit func(string, string)){
+						emit("Beijing", "China")
+						emit("London", "United Kingdom")
+						emit("San Francisco", "United States")
+						emit("Singapore", "Singapore")
+						emit("Sydney", "Australia")
+					},
+					beam.Impulse(s)),
+				persons: beam.Create(s,
+					task.Person{Name: "Henry", City: "Singapore"},
+					task.Person{Name: "Jane", City: "San Francisco"},
+					task.Person{Name: "Lee", City: "Beijing"},
+					task.Person{Name: "John", City: "Sydney"},
+					task.Person{Name: "Alfred", City: "London"},
+				),
+			},
+			want: []interface{}{
+				task.Person{
+					Name:    "Henry",
+					City:    "Singapore",
+					Country: "Singapore",
+				},
+				task.Person{
+					Name:    "Jane",
+					City:    "San Francisco",
+					Country: "United States",
+				},
+				task.Person{
+					Name:    "Lee",
+					City:    "Beijing",
+					Country: "China",
+				},
+				task.Person{
+					Name:    "John",
+					City:    "Sydney",
+					Country: "Australia",
+				},
+				task.Person{
+					Name:    "Alfred",
+					City:    "London",
+					Country: "United Kingdom",
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		got := task.ApplyTransform(s, tt.args.persons, tt.args.citiesAndCountriesKV)
+		passert.Equals(s, got, tt.want...)
+		if err := ptest.Run(p); err != nil {
+			t.Error(err)
+		}
+	}
+}

learning/katas/go/Core Transforms/Side Input/lesson-info.yaml

@@ -0,0 +1,21 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#  http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+content:
+- Side Input

learning/katas/go/Core Transforms/Side Input/lesson-remote-info.yaml

@@ -0,0 +1,3 @@
+id: 370175
+update_date: Mon, 22 Jun 2020 02:53:12 UTC
+unit: 355616

learning/katas/go/Core Transforms/section-info.yaml

@@ -24,3 +24,4 @@ content:
 - Combine
 - Flatten
 - Partition
+- Side Input

learning/katas/go/course-remote-info.yaml

@@ -1,2 +1,2 @@
 id: 70387
-update_date: Wed, 10 Jun 2020 21:21:51 UTC
+update_date: Tue, 16 Jun 2020 05:03:01 UTC