Dataflow は Apache Beam のマネージドサービスです。大規模データの分散処理が可能になります。簡単な使い方を記載します。

Dataflow テンプレートを用いる場合

GCP が提供するデータ処理のテンプレートを利用すると、Apache Beam の使い方を把握していなくても Dataflow を利用できます。

ここでは簡単な例として、Cloud Storage (GCS) 上の CSV ファイルを、DLP API によって de-identification して、BigQuery に書き出すデータフローのテンプレートを使ってみます。

DLP テンプレートの準備

PERSON_NAME を検出する inspection-template を作ってみます。

de-identification-template は、CSV の特定のカラムのみ匿名化したいため、Record を選択しています。

匿名化したいカラムのヘッダーは pii であるとします。pii カラムのデータについて、検出された箇所を info type 名で置換するように設定しています。

「テスト」タブで動作検証できます。inspection-template を指定することを忘れないようにします。pii 列の PERSON_NAME が匿名化されていることが分かります。

BigQuery データセットの準備

DLP で匿名化したデータを格納する BigQuery のデータセットを準備します。

bq mk mydataset_20210501

GCS バケットの準備

DLP で匿名化したいデータを格納する GCS バケットを準備します。

gsutil mb gs://mybucket-20210501-4

サービスアカウントの準備

GCP の Compute Engine サービス等を有効化すると、以下のようなサービスアカウントが作成されます。

{project-number}-compute@developer.gserviceaccount.com

既定では Dataflow はこのサービスアカウントを用いて、必要な VM を起動したりします。プロジェクトの Editor ロールが付与された状態で作成されますが、更に Cloud Dataflow Service Agent ロールを付与しておきます。

Dataflow ジョブの作成

Dataflow template に Data Masking/Tokenization from Cloud Storage to BigQuery (using Cloud DLP) を指定します。

GCS、Bigquery、DLP de-identification template の情報を指定します。

DLP inspection template の指定は必須ではありませんが、今回 PERSON_NAME 以外は検出してほしくないため、指定します。更に、Dataflow が起動する VM のタイプや台数、Dataflow が利用するサービスアカウント等を指定します。

ストリーミングタイプの Dataflow Job が作成されたことが確認できます。

指定した台数の VM が起動しています。

GCS への CSV ファイルアップロード

Dataflow Job のパラメータとして指定した GCS のパスに、CSV ファイルをアップロードします。

gsutil cp pii-sample.csv gs://mybucket-20210501-4/input/

pii-sample.csv

pii,non_pii
My name is Mike,My name is Mike
My phone number is 090-1234-1234,My phone number is 090-1234-1234

pii カラムの PERSON_NAME が匿名化されたことが確認できます。

Apache Beam 2.x Java SDK の利用例

先程用いた DLP による匿名化のテンプレートは、De-identification and re-identification of PII in large-scale datasets using Cloud DLP に記載されているユースケースの一つをサポートしています。

Dataflow テンプレートをそのまま利用できない場合は、Apache Beam SDK を用いて開発する必要があります。言語は Java または Python がサポートされています。以下では、既存のテンプレートを編集してカスタマイズする場合を想定しています。

Java8 および Maven3 のインストール

Dataflow テンプレートはオープンソースとして GitHub で公開されています。

git clone git@github.com:GoogleCloudPlatform/DataflowTemplates.git

テンプレートは Java で記述されています。Java8 および Maven3 をローカル環境にインストールして、以下のようにビルドできます。

mvn clean compile

Java15 等ではビルドできないことに注意します。

$ mvn --version
Apache Maven 3.8.1 (05c21c65bdfed0f71a2f2ada8b84da59348c4c5d)
Maven home: c:\Users\username\bin\apache-maven-3.8.1\bin\..
Java version: 1.8.0_291, vendor: Oracle Corporation, runtime: C:\Program Files\Java\jdk1.8.0_291\jre
Default locale: ja_JP, platform encoding: MS932
OS name: "windows 10", version: "10.0", arch: "amd64", family: "windows"

DLPTextToBigQueryStreaming.java をテストするには以下のようにします。

mvn test -Dtest=DLPTextToBigQueryStreamingTest

実行結果の例

[INFO] Results:
[INFO] 
[INFO] Tests run: 1, Failures: 0, Errors: 0, Skipped: 0
[INFO] 
[INFO] ------------------------------------------------------------------------
[INFO] BUILD SUCCESS
[INFO] ------------------------------------------------------------------------
[INFO] Total time:  44.991 s
[INFO] Finished at: 2021-05-02T17:29:09+09:00
[INFO] ------------------------------------------------------------------------

BigQuery データセットの準備

DLP で匿名化したデータを格納する BigQuery のデータセットを準備します。

bq mk mydataset_20210501

サービスアカウントの準備

ローカルPC で DLP API を実行して、結果を BigQuery に格納するためのサービスアカウントを準備します。

IntelliJ IDEA の設定

開発環境として IntelliJ IDEA を利用する場合は、DLPTextToBigQueryStreaming.java を以下のように実行できます。

「File → Project Structure...」で JDK8 を指定することを忘れないように注意します。

ローカル PC でのビルドおよび実行

ビルドした結果を GCS にアップロードして Dataflow Job を作成することでもソースコードの検証は行えます。そうではなく、ローカルPC で直接 Java を実行するには以下のようにします。

「Run → Run... → Edit Configurations...」

Environment variables に GOOGLE_APPLICATION_CREDENTIALS を設定して、サービスアカウントの鍵ファイルを指すようにしておきます。更に、Program arguments に必要な情報を指定しておきます:

--project=my-project-20210328
--dlpProjectId=my-project-20210328
--inputFilePattern=C:\Users\username\Desktop\pii-sample.csv
--inspectTemplateName=projects/my-project-20210328/locations/global/inspectTemplates/my-inspection-template-20210501
--deidentifyTemplateName=projects/my-project-20210328/locations/global/deidentifyTemplates/my-deidentification-template-20210501
--datasetName=mydataset_20210501

実行結果の例

BigQuery にも正しく格納されました。

Apache Beam 2.x Python SDK の利用例

インストール

Python から Apache Beam を利用するためには apache-beam をインストールします。setup.py に記載のとおり、GCP API を利用するための GCP_REQUIREMENTS も依存モジュールとしてインストールするように指定します。

python3 -m pip install apache-beam[gcp]

簡単なサンプル

my-dataflow-1.py

#!/usr/bin/python
# -*- coding: utf-8 -*-

from argparse import ArgumentParser

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.options.pipeline_options import SetupOptions
from apache_beam.io import ReadFromText

import logging
logging.getLogger().setLevel(logging.INFO)


def Main():

    parser = ArgumentParser()

    parser.add_argument(
        '--input', type=str,
        default='gs://dataflow-samples/shakespeare/kinglear.txt',
        help='Input file to process [default=%(default)s]')

    knownArgs, pipelineArgs = parser.parse_known_args()

    pipelineOptions = PipelineOptions(pipelineArgs)
    pipelineOptions.view_as(SetupOptions).save_main_session = True

    # パイプラインは with ブロックを抜けた時に実行されます。
    with beam.Pipeline(options=pipelineOptions) as p:
        lines = p | 'read from text' >> ReadFromText(knownArgs.input)
        lines | beam.Map(print)


if __name__ == '__main__':
    Main()

Apache Beam SDK には runner という概念があります。既定では DirectRunner が使われてローカル実行されます。

python3 my-dataflow-1.py --input ~/Desktop/pii-sample.csv

出力例

INFO:root:Missing pipeline option (runner). Executing pipeline using the default runner: DirectRunner.
...

pii,non_pii
My name is Mike,My name is Mike
My phone number is 090-1234-1234,My phone number is 090-1234-1234

DataflowRunner を指定して、その他の必要なオプションを pipelineArgs に渡すことで、GCP Dataflow で処理を実行できます。

export GOOGLE_APPLICATION_CREDENTIALS='C:\Users\username\Downloads\my-project-20210328-04084aee5ee7.json'

python3 my-dataflow-1.py \
--project my-project-20210328 \
--region us-central1 \
--runner DataflowRunner \
--staging_location gs://mybucket-20210501-4/staging \
--temp_location gs://mybucket-20210501-4/temp

作成された Dataflow Job

その他

Apache Beam で分散処理を行う際に必要となる関数の使い方は、以下のようなページに記載されています。

• CombinePerKey
• ParDo
• wordcount.py
• apache_beam.io.textio.ReadFromText

Apache Beam SDK で作成したパイプラインの利用方法について

Apache Beam SDK で DirectRunner を用いてローカル開発したパイプラインは、DataflowRunner で GCP 上で実行できます。

実行方法としては、以下の二つがあります。

• 本ページで行ったように GCS に staging するのと同時に Job を作成する方法。
• GCP で用意されているテンプレートのように GCS へのアップロードのみを行い、Dataflow Job は必要な時に Cloud Console (Web UI) 等から作成する方法。

更に後者については、Classic templates と Flex Templates の二つの選択肢があります。