はじめに

GCPのデータパイプラインを構築するプロダクトである、Dataflowを利用して
DataStoreにデータを登録する時のユーザー定義関数について共有します。

初めての方は、以下記事を前提にしておりますので、参照をお願いします。
https://qiita.com/sasuke9097/items/c3538933060f18b4c199

前提

Dataflowを利用してDataStoreにデータが登録できること
簡単なJavaScriptを記述できること

できるようになること

DataFlowのUDFを利用したデータの加工処理ができるようになる

ユーザー定義関数の概要

GoogleではUDF(User Define Function)と呼びます。
入力から出力の間に加工を施したり、登録するデータにフィルターを
かけることができます。

ユーザー定義関数実行までの流れ

ユーザー定義関数実行には、以下の手順を踏みます。

ユーザー定義関数の作成
作成したユーザー定義関数をGCSに配置
JOB実行時のパラメータに配置したユーザー定義関数を指定

ユーザー定義関数の作成

ユーザー定義関数を作成します。
ルールは、JOBから呼び出すときの関数を作成すること。
名前は任意です。今回は「transform」です。
プロパティを追加、変更、削除しています。
DataStoreの型によって、StringValueの部分は変わります。

my_function.js

function transform(inJson){
    // JSON文字列をオブジェクトに変換
    var json = JSON.parse(inJson);

    // プロパティ値を変更
    json.properties.s.stringValue = 'transform';

    // プロパティを追加
    json.properties.w = addStringValue('new prop');

    // プロパティを削除
    delete json.properties.t;

    // オブジェクトをJSON文字列に変換
    return JSON.stringify(json);
}

function addStringValue(value){
    newprop = new Object();
    newprop.stringValue = value;
    return newprop;
}

作成したユーザー定義関数をGCSに配置

作成したユーザー定義関数はGCSの任意のフォルダにアップロードしてください。

JOB実行時のパラメータに配置したユーザー定義関数を指定

JOB実行時に、パラメータで実行するユーザー定義関数を指定してください。
指定するのは、以下の二つです。
1. ユーザー定義関数ファイルのパス(javascriptTextTransformGcsPath)
2. ユーザー定義関数名(javascriptTextTransformFunctionName)

gcloud dataflow jobs run testText2Datastore
--gcs-location gs://dataflow-templates-asia-northeast1/latest/GCS_Text_to_Datastore
--region asia-northeast1
--staging-location gs://test-20201003/temp/
--parameters 
textReadPattern=gs://test-20201003/out-00000-of-00001.json,
datastoreWriteProjectId=ace-charter-291406,
errorWritePath=gs://test-20201003/error/failure.txt,
javascriptTextTransformGcsPath=gs://test-20201003/my_function.js,
javascriptTextTransformFunctionName=transform

コマンドの実行

あとは、今まで通り実行してください。

実行結果

指定したプロパティが、追加、変更、削除されています。

最後に

今回は、追加、変更、削除をしました。
プロパティの削除はなかなか無いとは思います。
プロパティの変更は、よくあると思います。
プロパティの追加は、例えば処理済みフラグや、実行時刻の追加等に使えるのでは
と思いました。
次回は、ユーザー定義関数を利用したフィルターについてお伝えしようと思います。

【GCP】DataFlowを利用してDataStoreにデータを登録(ユーザー定義関数)