ApacheBeam-Tutorial(GoLangVersion)

Apacheのアプリを調べてたところ、AzureDataFactoryのようなものを発見しました。
こちらになります。

https://beam.apache.org/

SaaSではなく、コードベースになっているデータインテグレーションフレームワークになります。
まだ詳しくない状況ですが、GUI等がないので、直感的にわかりにくいです。
これを使う利点として、他のApacheのアプリと結合しやすい点ぐらいしか思い浮かばないのです。

早速ですが、上記のページの公式ドキュメントについて説明になります。

## 使うテストデータ
Apache Beamのところから適当に取得しました。
ここはセットしたいデータで問題ないです。
```text
The SDK-provided Count transform is a generic transform that takes a PCollection of any type, and returns a PCollection of key/value pairs. Each key represents a unique element from the input collection, and each value represents the number of times that key appeared in the input collection.
In this pipeline, the input for Count is the PCollection of individual words generated by the previous ParDo, and the output is a PCollection of key/value pairs where each key represents a unique word in the text and the associated value is the occurrence count for each.
```

## 使うソースコード

公式のソースコードを使います。が、Goバージョンは現在一部の障害があります。
その点について最後に取り上げます。

```Golang
package main

//https://beam.apache.org/get-started/quickstart-go/
import (
	"context"
	"flag"
	"fmt"
	"regexp"

	"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/runners/direct"
	"github.com/apache/beam/sdks/v2/go/pkg/beam/transforms/stats"

	_ "github.com/apache/beam/sdks/v2/go/pkg/beam/io/filesystem/local"
)

var (
	input  = flag.String("input", "data/*", "File(s) to read.")
	output = flag.String("output", "outputs/wordcounts.txt", "Output filename.")
)

var wordRE = regexp.MustCompile(`[a-zA-Z]+('[a-z])?`)

func main() {
	flag.Parse()

	beam.Init()

	pipeline := beam.NewPipeline()
	root := pipeline.Root()

	lines := textio.Read(root, *input)
	words := beam.ParDo(root, func(line string, emit func(string)) {
		for _, word := range wordRE.FindAllString(line, -1) {
			emit(word)
		}
	}, lines)

	counted := stats.Count(root, words)

	formatted := beam.ParDo(root, func(word string, count int) string {
		return fmt.Sprintf("%s: %v", word, count)
	}, counted)

	//https://github.com/apache/beam/issues/21998
	textio.Write(root, *output, formatted)

	direct.Execute(context.Background(), pipeline)
}

```

1. 実行の引数として、`--input`と`--output`を指定します。
1. Beamを初期化します。
    ```	Golang
    beam.Init()
    ```
1. PipeLineを定義します。
    ```	Golang
	pipeline := beam.NewPipeline()
	root := pipeline.Root()
    ```
1. Inputデータを指定します。
    ```	Golang
	lines := textio.Read(root, *input)
    ```
1. Inputデータから各単語単位で分割するジョブです。
    ```	Golang
	words := beam.ParDo(root, func(line string, emit func(string)) {
		for _, word := range wordRE.FindAllString(line, -1) {
			emit(word)
		}
	}, lines)

    ```
1. 各単語が使われた回数を集計します。
    ```	Golang
	counted := stats.Count(root, words)
    ```

1. 結果を出力するためのフォマットを設定します。
    ```	Golang
	formatted := beam.ParDo(root, func(word string, count int) string {
		return fmt.Sprintf("%s: %v", word, count)
	}, counted)
    ```
1. 結果をファイルへ書き記みます。
    ```	Golang
	textio.Write(root, *output, formatted)
    ```

1. 設定したバッチを実行します。
    ```	Golang
	direct.Execute(context.Background(), pipeline)
    ```

## 結果
Goで実行することになると、無限ループに入ることになります。Commnad+Cで抜けられます。
![スクリーンショット 2022-07-10 10.14.46.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1089747/2f156202-09f8-e140-58b5-cf855854e648.png)

ソースコードのどこが悪いかかと最初に思うかもしれませんが、どうやら別の問題のようです。
https://github.com/apache/beam/issues/21998
こちらの障害報告のチケットがありました。結果をファイルに書き込む際に無限ループに入ることになってそうです。
すぐ解決は出来無さそうですので原因について調べてます。そこはまた別の記事で扱います。
無限ループの途中に抜けた際の結果ですが、下記のようになります。
![スクリーンショット 2022-07-10 10.15.42.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1089747/5fa6282c-dbfb-677b-807f-27358caa0dc2.png)
想定通り、各単語が使われた頻度が集計されておリます。

## 終わり
今回はBeamのTutorialを行い、動作確認レベルで実演してみました。Pipelineで行う操作についてソースコードで書くことでカスタマイズできるのでBatchシステムでも使えそうです。また、Triggerの設定でバッチが動くタイミングを設定もできると記載があります。上記の障害を調べながらApache BeamのLife Cycleや内部Systemについて少し気になったので次回はそこの確認をしたいと思います。