Cloud Dataflow for JavaをAppEngine for Goから実行する

Cloud Dataflow Template

従来のCloud Dataflowはバッチモードで起動する場合、どこかからデプロイする必要がありました。
しかし、Dataflow Templateを使えば、事前にCloud Storageに配置しておき、任意のタイミングで実行することができます。

Dataflow Templateの作成

Dataflow Templateは以下のように作成します。

mvn compile exec:java -Dexec.mainClass=org.sinmetal.flow.BigQueryToDatastore \
              -Dexec.args="--project={your GCP ProjectID} \
              --stagingLocation=gs://{your bucket} \
              --dataflowJobFile=gs://{your output Dataflow Template Path} \
              --runner=TemplatingDataflowPipelineRunner"

Parameter

Dataflow Template作成時に以下のようなパラメータを渡します。
dataflowJobFile がTemplateの名前のようなもので、Templateを実行する時に指定します。

• project : GCP ProjectID
• stagingLocation : Dataflowが利用するファイルを置くGCS Bucket
• dataflowJobFile : Dataflow Templateの設定ファイルを作成するGCS Path

Dataflow Templateの実行

Dataflow Templateを実行する方法は3つあります。

• gcloud sdkから実行
• REST APIから実行
• Client Libraryから実行

今回は、gcloud sdk からの実行と、 Client Libraryからの実行 について説明します。

gcloud sdkから実行

gcloud sdkで実行する場合は以下のようになります。

{job_name}は起動したDataflowのJobのNameとして設定される任意の値です。

gcloud beta dataflow jobs run {job_name} \
        --gcs-location gs://{your Dataflow Template Path}

Client Libraryから実行

今回は、App Engine Goから実行しました。
以下のような感じで、Client LibraryからDataflowのAPIを叩きます。

import (
	"fmt"
	"io"
	"net/http"

	"golang.org/x/oauth2"
	"golang.org/x/oauth2/google"

	"google.golang.org/appengine"
	"google.golang.org/appengine/log"
	"google.golang.org/appengine/urlfetch"
)

const ProjectID = "your project id"

func init() {
	http.HandleFunc("/cron/start", handler)
}

func handler(w http.ResponseWriter, r *http.Request) {
	ctx := appengine.NewContext(r)

	client := &http.Client{
		Transport: &oauth2.Transport{
			Source: google.AppEngineTokenSource(ctx, "https://www.googleapis.com/auth/cloud-platform"),
			Base:   &urlfetch.Transport{Context: ctx},
		},
	}

	res, err := client.Post(fmt.Sprintf("https://dataflow.googleapis.com/v1b3/projects/%s/templates", ProjectID), "application/json", r.Body)
	if err != nil {
		log.Errorf(ctx, "ERROR dataflow: %s", err)
		w.WriteHeader(http.StatusInternalServerError)
		return
	}

	_, err = io.Copy(w, res.Body)
	if err != nil {
		log.Errorf(ctx, "ERROR Copy API response: %s", err)
		w.WriteHeader(http.StatusInternalServerError)
		return
	}
	w.WriteHeader(res.StatusCode)
}

今回はJSONをぶん投げて、パラメータを渡しています。

{
	"jobName": "{job_name}",
	"gcsPath": "gs://{your Dataflow Template Path}",
	"environment": {
		"tempLocation": "gs://{your bucket}",
		"zone": "us-central1-f"
	}
}

#おわり

DataflowはAppEngineの苦手なバッチ処理をやってくれるフルマネージドサービスです。
料金も案外安いので、AppEngineでごりごりTaskQueueでやってたようなCloud StorageからCSVを読み込んで、Datastoreへ書き込むみたいなことは、Dataflowでやった方が簡単かもしれません。
バッチ処理の起動タイミングも、Dataflow Templateで解決しました。
これからは、もりもりDataflowの知見をためていきましょう！

#Next

Cloud Dataflow for Javaを実行時パラメータ付きでAppEngine for Goから実行する

#for Python

Cloud Dataflow(Python)をAppEngineから実行する