Skip to content

Commit

Permalink
[Tour of Beam] Learning content for "Final challenge" module (#26861)
Browse files Browse the repository at this point in the history
* add final challenge

* correct examples

* changge

* correct golang

* correct example hint

* correct examples and add golang example

* delete whitespace

* delete whitespace

* add file tag

* correct final challenge

* fixing incorrect tags and names

* minor formatting

* fixing example formatting

* correct challenge

* correct whitespace

* change final-challenge-2

* correct imports

* fix splittable unit id

* formatting

* format change

* formatting

* fixing template

* temp fix of url

* backend urls fix

* restore backend urls

* change

* remove window

---------

Co-authored-by: mende1esmende1es <mende1esmende1es@gmail.cp>
Co-authored-by: Oleh Borysevych <oleg.borisevich@akvelon.com>
  • Loading branch information
3 people authored Jun 26, 2023
1 parent ca806dc commit 1785a98
Show file tree
Hide file tree
Showing 33 changed files with 34,161 additions and 2 deletions.
3 changes: 2 additions & 1 deletion learning/tour-of-beam/learning-content/content-info.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -31,4 +31,5 @@ content:
- io
- splittable-dofn
- cross-language

- final-challenge

Original file line number Diff line number Diff line change
@@ -0,0 +1,27 @@
<!--
Licensed 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.
-->
### Final challenge 1

You’re given a csv file with purchase transactions. Write a Beam pipeline to prepare a report every 30 seconds. The report needs to be created only for transactions where quantity is more than 20.

Report should consist of two files named "**price_more_than_10.txt**" and "**price_less_than_10.txt**":

* Total transactions amount grouped by **ProductNo** for products with **price** greater than 10
* Total transactions amount grouped by **ProductNo** for products with **price** less than 10

Example rows from input file:

| TransactionNo | Date | ProductNo | ProductName | Price | Quantity | CustomerNo | Country |
|---------------|-----------|-----------|-------------------------------------|-------|----------|------------|----------------|
| 581482 | 12/9/2019 | 22485 | Set Of 2 Wooden Market Crates | 21 | 47 | 17490 | United Kingdom |
| 581475 | 12/9/2019 | 22596 | Christmas Star Wish List Chalkboard | 10.65 | 36 | 13069 | United Kingdom |
| 581475 | 12/9/2019 | 23235 | Storage Tin Vintage Leaf | 11.53 | 12 | 13069 | United Kingdom |

Large diffs are not rendered by default.

Original file line number Diff line number Diff line change
@@ -0,0 +1,65 @@
/*
* 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.
*/

// beam-playground:
// name: FinalChallenge1
// description: Final challenge 1.
// multifile: true
// files:
// - name: input.csv
// context_line: 54
// categories:
// - Quickstart
// complexity: ADVANCED
// tags:
// - hellobeam

package main

import (
"context"
"github.com/apache/beam/sdks/v2/go/pkg/beam"
"github.com/apache/beam/sdks/v2/go/pkg/beam/io/textio"
"github.com/apache/beam/sdks/v2/go/pkg/beam/x/beamx"
"log"
)

type Transaction struct {
ID int64
Date string
ProductID string
ProductName string
Price float64
Quantity int64
CustomerID int64
Country string
}

func main() {
beam.Init()
p := beam.NewPipeline()
s := p.Root()

file := textio.Read(s, "input.csv")

textio.Write(s, "price_less_than_10.txt", file)

if err := beamx.Run(context.Background(), p); err != nil {
log.Fatalf("Failed to execute job: %v", err)
}
}

Large diffs are not rendered by default.

Original file line number Diff line number Diff line change
@@ -0,0 +1,150 @@
/*
* 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.
*/

// beam-playground:
// name: FinalSolution1
// description: Final challenge solution 1.
// multifile: true
// files:
// - name: input.csv
// context_line: 54
// categories:
// - Quickstart
// complexity: ADVANCED
// tags:
// - hellobeam

package main

import (
"context"
"fmt"
"github.com/apache/beam/sdks/v2/go/pkg/beam"
"github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/window"
"github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/window/trigger"
"github.com/apache/beam/sdks/v2/go/pkg/beam/io/textio"
"github.com/apache/beam/sdks/v2/go/pkg/beam/transforms/filter"
"github.com/apache/beam/sdks/v2/go/pkg/beam/transforms/stats"
"github.com/apache/beam/sdks/v2/go/pkg/beam/x/beamx"
"log"
"strconv"
"strings"
"time"
)

type Transaction struct {
ID int64
Date string
ProductID string
ProductName string
Price float64
Quantity int64
CustomerID int64
Country string
}

func main() {
ctx := context.Background()

beam.Init()
p := beam.NewPipeline()
s := p.Root()

file := textio.Read(s, "input.csv")

transactions := getTransactions(s, file)

trigger := trigger.AfterEndOfWindow().
EarlyFiring(trigger.AfterProcessingTime().
PlusDelay(5 * time.Second)).
LateFiring(trigger.Repeat(trigger.AfterCount(1)))

fixedWindowedItems := beam.WindowInto(s, window.NewFixedWindows(30*time.Second), transactions,
beam.Trigger(trigger),
beam.AllowedLateness(30*time.Minute),
beam.PanesDiscard(),
)

filtered := filtering(s, fixedWindowedItems)

result := getPartition(s, filtered)

biggerThan10 := sumCombine(s, mapIdWithPrice(s, result[0]))
textio.Write(s, "price_more_than_10.txt", convertToString(s, biggerThan10))

smallerThan10 := sumCombine(s, mapIdWithPrice(s, result[1]))
textio.Write(s, "price_less_than_10.txt", convertToString(s, smallerThan10))

if err := beamx.Run(ctx, p); err != nil {
log.Fatalf("Failed to execute job: %v", err)
}
}

func getTransactions(s beam.Scope, input beam.PCollection) beam.PCollection {
return beam.ParDo(s, func(line string, emit func(transaction Transaction)) {
csv := strings.Split(line, ",")

if csv[0] != "TransactionNo" {
id, _ := strconv.ParseInt(csv[0], 10, 64)
price, _ := strconv.ParseFloat(csv[4], 64)
quantity, _ := strconv.ParseInt(csv[5], 10, 64)
customerID, _ := strconv.ParseInt(csv[6], 10, 64)
emit(Transaction{
ID: id,
Date: csv[1],
ProductID: csv[2],
ProductName: csv[3],
Price: price,
Quantity: quantity,
CustomerID: customerID,
Country: csv[7],
})
}
}, input)
}

func getPartition(s beam.Scope, input beam.PCollection) []beam.PCollection {
return beam.Partition(s, 2, func(element Transaction) int {
if element.Price >= 10 {
return 0
}
return 1
}, input)
}

func convertToString(s beam.Scope, input beam.PCollection) beam.PCollection {
return beam.ParDo(s, func(product string, sum float64, emit func(string)) {
emit(fmt.Sprint("product: ", product, " , sum: ", sum))
}, input)
}

func filtering(s beam.Scope, input beam.PCollection) beam.PCollection {
return filter.Include(s, input, func(element Transaction) bool {
return element.Quantity >= 20
})
}

func mapIdWithPrice(s beam.Scope, input beam.PCollection) beam.PCollection {
return beam.ParDo(s, func(element Transaction, emit func(string, float64)) {
emit(element.ProductID, element.Price)
}, input)
}

func sumCombine(s beam.Scope, input beam.PCollection) beam.PCollection {
return stats.SumPerKey(s, input)
}
Loading

0 comments on commit 1785a98

Please sign in to comment.