JupyterLab ノートブックと Apache Beam インタラクティブ ランナーを使うと、「入力、評価、出力」ループ（REPL）のワークフローで反復するパイプラインの開発、パイプライン グラフの検査、個々の PCollection の解析ができます。これらの Apache Beam ノートブックは、Vertex AI Workbench ユーザー管理ノートブックで提供されています。これは、最新のデータ サイエンスと機械学習フレームワークがプリインストールされたノートブック仮想マシンをホストするマネージド サービスです。
このガイドでは、Apache Beam ノートブックを導入することで実現する機能を中心に取り上げますが、Apache Beam ノートブックのビルド方法については説明しません。Apache Beam の詳細については、Apache Beam プログラミング ガイドをご覧ください。
始める前に
- Google Cloud アカウントにログインします。Google Cloud を初めて使用する場合は、アカウントを作成して、実際のシナリオでの Google プロダクトのパフォーマンスを評価してください。新規のお客様には、ワークロードの実行、テスト、デプロイができる無料クレジット $300 分を差し上げます。
-
Google Cloud Console の [プロジェクト セレクタ] ページで、Google Cloud プロジェクトを選択または作成します。
-
Cloud プロジェクトに対して課金が有効になっていることを確認します。詳しくは、プロジェクトで課金が有効になっているかどうかを確認する方法をご覧ください。
-
Compute Engine, Notebooks API を有効にします。
-
Google Cloud Console の [プロジェクト セレクタ] ページで、Google Cloud プロジェクトを選択または作成します。
-
Cloud プロジェクトに対して課金が有効になっていることを確認します。詳しくは、プロジェクトで課金が有効になっているかどうかを確認する方法をご覧ください。
-
Compute Engine, Notebooks API を有効にします。
このガイドの完了後に作成したリソースを削除すれば、今後課金は発生しません。詳しくは、クリーンアップをご覧ください。
Apache Beam ノートブック インスタンスを起動する
Google Cloud コンソールで、Dataflow の [ワークベンチ] ページに移動します。
[ユーザー管理のノートブック] タブが表示されていることを確認します。
ツールバーで [ 新しいノートブック] をクリックします。
[Apache Beam] > [Withoug GPUs] を選択します。
（省略可）GPU 上でノートブックを実行する場合は、[Apache Beam] > [With 1 NVIDIA Tesla T4] を選択します。
[新しいノートブック] ページで、ノートブック VM のサブネットワークを選択します。
GPU を備えたノートブック インスタンスを作成する場合は、[新しいノートブック] ページで [NVIDIA GPU ドライバを自動的にインストールする] を選択します。
（省略可）カスタム ノートブック インスタンスを設定する場合は、[詳細オプション] をクリックします。インスタンス プロパティのカスタマイズについて詳しくは、特定のプロパティでユーザー管理のノートブック インスタンスを作成するをご覧ください。
[作成] をクリックします。
リンクがアクティブになったら、[JupyterLab を開く] をクリックします。Vertex AI Workbench が、新しい Apache Beam ノートブック インスタンスを作成します。
依存関係をインストールする（省略可）
Apache Beam ノートブックには、Apache Beam と Google Cloud コネクタの依存関係がすでにインストールされています。サードパーティのライブラリに依存するカスタム コネクタや PTransform がパイプラインに含まれている場合は、ノートブック インスタンスを作成した後にインストールできます。詳細については、ユーザー管理ノートブックのドキュメントに関する依存関係のインストールをご覧ください。
Apache Beam ノートブックのスタートガイド
ユーザー管理のノートブック インスタンスを開くと、Examples フォルダにサンプル ノートブックが表示されます。現在利用できるサンプルは、次のとおりです。
- Word Count
- Streaming Word Count
- Streaming NYC Taxi Ride Data
- Dataflow Word Count
- ノートブックでの Apache Beam SQL
- 大規模でインタラクティブな Flink
- Apache Beam で GPU を使用する
- データを可視化する
Apache Beam の基礎知識を説明する追加のチュートリアルが [Tutorials] フォルダにあります。現在利用できるサンプルは、次のとおりです。
- 基本オペレーション
- 要素ごとのオペレーション
- 集計
- Windows
- IO オペレーション
- ストリーミング
これらのノートブックには、Apache Beam のコンセプトと API の使用方法の理解を助ける説明テキストとコメント付きのコードブロックが含まれています。また、チュートリアルでは、学習したコンセプトを実践で練習できます。
ノートブック インスタンスを作成する
[File] > [New] > [Notebook] に移動し、Apache Beam 2.22 以降のカーネルを選択します。
Apache Beam はノートブック インスタンスにインストールされているため、ノートブックには interactive_runner モジュールと interactive_beam モジュールが含まれます。
import apache_beam as beam
from apache_beam.runners.interactive.interactive_runner import InteractiveRunner
import apache_beam.runners.interactive.interactive_beam as ib
ノートブックで他の Google API を使用している場合は、次の import ステートメントを追加します。
from apache_beam.options import pipeline_options
from apache_beam.options.pipeline_options import GoogleCloudOptions
import google.auth
インタラクティブ オプションを設定する
以下は、InteractiveRunner が無制限のソースからデータを記録するまでの時間を設定しています。この例では、期間は 10 分に設定されています。
ib.options.recording_duration = '10m'
また、recording_size_limit プロパティを使用して、無制限のソースの記録サイズ制限（バイト単位）を変更することもできます。
# Set the recording size limit to 1 GB.
ib.options.recording_size_limit = 1e9
その他のインタラクティブ オプションについては、interactive_beam.options クラスをご覧ください。
パイプラインを作成する
InteractiveRunner オブジェクトを使用して、パイプラインを初期化します。
options = pipeline_options.PipelineOptions()
# Set the pipeline mode to stream the data from Pub/Sub.
options.view_as(pipeline_options.StandardOptions).streaming = True
p = beam.Pipeline(InteractiveRunner(), options=options)
データの読み取りと可視化
次の例は、指定された Pub/Sub トピックにサブスクリプションを作成し、そのサブスクリプションから読み取りを行う Apache Beam パイプラインを示しています。
words = p
| "read" >> beam.io.ReadFromPubSub(topic="projects/pubsub-public-data/topics/shakespeare-kinglear")
パイプラインは、ソースからウィンドウごとに単語をカウントします。固定されたウィンドウを作成し、各ウィンドウの長さを 10 秒に設定します。
windowed_words = (words
| "window" >> beam.WindowInto(beam.window.FixedWindows(10)))
データがウィンドウ処理されると、ウィンドウごとに単語をカウントします。
windowed_word_counts = (windowed_words
| "count" >> beam.combiners.Count.PerElement())
show() メソッドは、作成された PCollection をノートブックに可視化します。
ib.show(windowed_word_counts, include_window_info=True)
n と duration の 2 つのオプション パラメータを設定すると、show() から結果セットを絞り込むことができます。n を設定すると、結果セットが最大で n の要素数（20 など）を表示するように制限されます。n が設定されていない場合、デフォルトの動作として、ソースの記録が終了するまで最新の要素が一覧表示されます。duration を設定すると、結果セットはソース記録の開始から指定した秒数分のデータに制限されます。duration が設定されていない場合、デフォルトの動作として、記録が終了するまですべての要素が一覧表示されます。
両方のパラメータを設定した場合、いずれかがしきい値に達すると show() が停止します。次の例では、show() は記録されたソースからの最初の 30 秒に相当するデータに基づいて、計算された最大 20 個の要素を返します。
ib.show(windowed_word_counts, include_window_info=True, n=20, duration=30)
データの可視化を表示するには、visualize_data=True を show() メソッドに渡します。可視化には複数のフィルタを適用できます。次の可視化では、ラベルと軸でフィルタリングできます。
Apache Beam ノートブックの可視化では、Pandas DataFrame も役立ちます。次の例では、最初に単語を小文字に変換してから、各単語の頻度を計算します。
windowed_lower_word_counts = (windowed_words
| beam.Map(lambda word: word.lower())
| "count" >> beam.combiners.Count.PerElement())
collect() メソッドにより、Pandas DataFrame の出力を取得できます。
ib.collect(windowed_lower_word_counts, include_window_info=True)
Interactive Beam インスペクタでデータを可視化する
show() と collect() を絶え間なく呼び出すことで、PCollection のデータの取り込みを邪魔してしまう場合があります。特に出力が画面スペースの多くを占有し、ノートブックの操作が困難になることがあります。また、複数の PCollection を並べて比較し、変換が意図したとおりに機能するかどうかを検証することもできます。たとえば、1 つの PCollection が変換され、別の変換が生成される場合などです。これらのユースケースでは、Interactive Beam インスペクタのほうが便利です。
Interactive Beam インスペクタは、Apache Beam ノートブックにプリインストールされている JupyterLab 拡張機能 apache-beam-jupyterlab-sidepanel として提供されています。この拡張機能を使用すると、show() や collect() を明示的に呼び出すことなく、各 PCollection に関連付けられたパイプラインとデータの状態をインタラクティブに検査できます。
インスペクタを開く方法は 3 つあります。
JupyterLab の上部メニューバーで [
Interactive Beam] をクリックします。プルダウンで [Open Inspector] をクリックしてインスペクタを開きます。
ランチャー ページを使用します。ランチャー ページが開いていない場合は、[
File] -> [New Launcher] の順にクリックして開きます。ランチャー ページでInteractive Beamを探して [Open Inspector] をクリックし、インスペクタを開きます。
コマンド パレットを使用します。JupyterLab の上部メニューバーで [
View] -> [Activate Command Palette] の順にクリックします。ポップアップでInteractive Beamを検索して、拡張機能のすべてのオプションを一覧表示します。[Open Inspector] をクリックしてインスペクタを開きます。
インスペクタが開くときに:
開いているノートブックが 1 つだけの場合、インスペクタは自動的にそのノートブックに接続します。
それ以外の場合は、ダイアログが表示され、カーネル（ノートブックが開いていない場合）または接続先のノートブック セッション（複数のノートブックが開いている場合）を選択できます。
開いたノートブックに複数のインスペクタを開き、ワークスペースでタブを自由にドラッグ＆ドロップすることでインスペクタを並べ替えることができます。
ノートブックでセルを実行すると、インスペクタ ページが自動的に更新されます。左側には、接続されたノートブックで定義されたパイプラインと PCollection が表示されます。PCollection は、所属するパイプラインごとに整理されます。ヘッダーのパイプラインをクリックすると、折りたたむことができます。
パイプラインと PCollection リスト内のアイテムをクリックすると、対応する可視化データがインスペクタの右側にレンダリングされます。
PCollection の場合、インスペクタは、追加のウィジェットを使用してデータをレンダリングします（データがまだ制限なしの PCollection に送信されている場合は動的にレンダリングします）。[
APPLY] ボタンをクリックすると、可視化が調整されます。
パイプラインである場合は、インスペクタにパイプラインのグラフが表示されます。
匿名パイプラインが存在することがあります。これはアクセス可能な PCollection を含むパイプラインですが、メイン セッションによって参照されなくなります。次に例を示します。
p = beam.Pipeline()
pcoll = p | beam.Create([1, 2, 3])
p = beam.Pipeline()
上記の例では、空のパイプライン p と、1 つの PCollection pcoll を含む匿名パイプラインが作成されます。匿名パイプラインには、引き続き pcoll.pipeline を使用してアクセスできます。
左側のパイプラインと PCollection のリストを切り替えると、可視化に使用するスペースを広げることができます。
パイプラインの記録ステータスについて
可視化だけでなく、describe を呼び出すことで、ノートブック インスタンス内の 1 つまたはすべてのパイプラインの記録ステータスを調べることもできます。
# Return the recording status of a specific pipeline. Leave the parameter list empty to return
# the recording status of all pipelines.
ib.recordings.describe(p)
describe() メソッドは、次の詳細情報を提供します。
- ディスク上のパイプラインのすべての記録の合計サイズ（バイト単位）
- バックグラウンドの記録ジョブの開始時間（Unix エポックからの秒数）
- バックグラウンドの記録ジョブの現在のパイプライン ステータス
- パイプラインの Python 変数
ノートブックに作成されたパイプラインから Dataflow ジョブを起動する
- （省略可）ノートブックを使用して Dataflow ジョブを実行する前に、カーネルを再起動し、すべてのセルを再実行して出力を確認します。この手順を省略すると、ノートブックの隠れた状態がパイプライン オブジェクトのジョブグラフに影響を及ぼす可能性があります。
- Dataflow API を有効にします。
次の import ステートメントを追加します。
from apache_beam.runners import DataflowRunnerパイプライン オプションを渡します。
# Set up Apache Beam pipeline options. options = pipeline_options.PipelineOptions() # Set the project to the default project in your current Google Cloud # environment. _, options.view_as(GoogleCloudOptions).project = google.auth.default() # Set the Google Cloud region to run Dataflow. options.view_as(GoogleCloudOptions).region = 'us-central1' # Choose a Cloud Storage location. dataflow_gcs_location = 'gs://<change me>/dataflow' # Set the staging location. This location is used to stage the # Dataflow pipeline and SDK binary. options.view_as(GoogleCloudOptions).staging_location = '%s/staging' % dataflow_gcs_location # Set the temporary location. This location is used to store temporary files # or intermediate results before outputting to the sink. options.view_as(GoogleCloudOptions).temp_location = '%s/temp' % dataflow_gcs_location # If and only if you are using Apache Beam SDK built from source code, set # the SDK location. This is used by Dataflow to locate the SDK # needed to run the pipeline. options.view_as(pipeline_options.SetupOptions).sdk_location = ( '/root/apache-beam-custom/packages/beam/sdks/python/dist/apache-beam-%s0.tar.gz' % beam.version.__version__)パラメータ値は調整できます。たとえば、
regionの値をus-central1から変更できます。DataflowRunnerを使用してパイプラインを実行します。これにより、Dataflow サービスでジョブが実行されます。runner = DataflowRunner() runner.run_pipeline(p, options=options)pは、パイプラインの作成のパイプライン オブジェクトです。
インタラクティブ ノートブックでこの変換を実行する方法の例については、ノートブック インスタンスの Dataflow Word Count ノートブックを参照してください。
あるいは、ノートブックを実行可能スクリプトとしてエクスポートし、生成された .py ファイルを前の手順で変更してから、Dataflow サービスにパイプラインをデプロイすることもできます。
ノートブックを保存する
作成したノートブックは、実行中のノートブック インスタンスにローカルに保存されます。開発中にノートブック インスタンスをリセットまたはシャットダウンした場合、これらの新しいノートブックは、/home/jupyter ディレクトリの下に作成される限り保持されます。ただし、ノートブック インスタンスが削除されると、それらのノートブックも削除されます。
ノートブックを後で使用できるようにするには、ワークステーションにローカルにダウンロードするか、GitHub に保存するか、別のファイル形式にエクスポートします。
ノートブックを追加の永続ディスクに保存する
さまざまなノートブック インスタンスでノートブックやスクリプトなどの処理を保持する場合は、Persistent Disk に保存します。
Persistent Disk を作成するか、アタッチします。手順に沿って
sshを使用してノートブック インスタンスの VM に接続し、開いている Cloud Shell でコマンドを発行します。Persistent Disk がマウントされているディレクトリ（
/mnt/myDiskなど）をメモします。ノートブック インスタンスの VM の詳細を編集し、
Custom metadataに次のエントリを追加します: キー -container-custom-params、値 --v /mnt/myDisk:/mnt/myDisk。
[保存] をクリックします。
これらの変更を更新するには、ノートブック インスタンスをリセットします。
リセット後にリンクがアクティブになったら、[JupyterLab を開く] をクリックします。JupyterLab UI が利用可能になるまでに時間がかかることがあります。UI が表示されたら、ターミナルを開いてコマンド
ls -al /mntを実行します。/mnt/myDiskディレクトリが表示されます。
これで、作業を /mnt/myDisk ディレクトリに保存できるようになりました。ノートブック インスタンスが削除されても、Persistent Disk はプロジェクトに存在し続けます。この Persistent Disk は、他のノートブック インスタンスにアタッチできます。
クリーンアップ
Apache Beam ノートブック インスタンスの使用が終了したら、ノートブック インスタンスをシャットダウンして、Google Cloud で作成したリソースをクリーンアップします。
高度な機能
ノートブック管理のクラスタ上のインタラクティブな FlinkRunner
ノートブックから本番環境サイズのデータをインタラクティブに操作するには、いくつかの汎用的なパイプライン オプションとともに FlinkRunner を使用することで、ノートブック セッションに対して長期的な Dataproc クラスタを管理し、Beam パイプラインを分散して実行するように指示できます。
前提条件
この機能を使用するには:
- Dataproc API を有効にする必要があります。
- ノートブック インスタンスを実行するサービス アカウントには、Dataproc の管理者ロールまたは編集者ロールが必要です。
- Beam バージョン 2.40.0 以上のノートブック カーネルを使用します。
構成
最小設定:
# Set a Cloud Storage bucket to cache source recording and PCollections.
# By default, the cache is on the notebook instance itself, but that does not
# apply to the distributed execution scenario.
ib.options.cache_root = 'gs://<CHANGE_ME>/flink'
# Define an InteractiveRunner that uses the FlinkRunner under the hood.
interactive_flink_runner = InteractiveRunner(underlying_runner=FlinkRunner())
options = PipelineOptions()
# Instruct the notebook that Google Cloud is used to run the FlinkRunner.
cloud_options = options.view_as(GoogleCloudOptions)
cloud_options.project = '<YOUR-PROJECT>'
（オプション）明示的なプロビジョニング
その他のオプション
# Change this if the pipeline needs to be executed in a different region
# than the default 'us-central1'. For example, to set it to 'us-west1':
cloud_options.region = 'us-west1'
# Explicitly provision the notebook-managed cluster.
worker_options = options.view_as(WorkerOptions)
# Provision 40 workers to run the pipeline.
worker_options.num_workers=40
# Use the default subnetwork.
worker_options.subnetwork='default'
# Choose the machine type for the workers.
worker_options.machine_type='n1-highmem-8'
# When working with non-official Beam releases, e.g. Beam built from source
# code, configure the environment to use a compatible released SDK container.
# If needed, build a custom container and use it (to learn more, see:
# https://beam.apache.org/documentation/runtime/environments/)
options.view_as(PortableOptions).environment_config = 'apache/beam_python3.7_sdk:2.41.0 or gcr.io/your_custom_container'
使用法
# The parallelism is applied to each step, so if your pipeline has 10 steps, you
# end up having 10 * 10 = 100 tasks scheduled that can be executed in parallel.
options.view_as(FlinkRunnerOptions).parallelism = 10
p_word_count = beam.Pipeline(interactive_flink_runner, options=options)
word_counts = (
p_word_count
| 'read' >> ReadWordsFromText('gs://apache-beam-samples/shakespeare/kinglear.txt')
| 'count' >> beam.combiners.Count.PerElement())
# The notebook session automatically starts and manages a cluster to execute
# your pipelines for execution with the FlinkRunner.
ib.show(word_counts)
# Interactively adjust the parallelism.
options.view_as(FlinkRunnerOptions).parallelism = 150
# The BigQuery read needs a Cloud Storage bucket as a temporary location.
options.view_as(GoogleCloudOptions).temp_location = ib.options.cache_root
p_bq = beam.Pipeline(runner=interactive_flink_runner, options=options)
delays_by_airline = (
p_bq
| 'Read Dataset from BigQuery' >> beam.io.ReadFromBigQuery(
project=project, use_standard_sql=True,
query=('SELECT airline, arrival_delay '
'FROM `bigquery-samples.airline_ontime_data.flights` '
'WHERE date >= "2010-01-01"'))
| 'Rebalance Data to TM Slots' >> beam.Reshuffle(num_buckets=1000)
| 'Extract Delay Info' >> beam.Map(
lambda e: (e['airline'], e['arrival_delay'] > 0))
| 'Filter Delayed' >> beam.Filter(lambda e: e[1])
| 'Count Delayed Flights Per Airline' >> beam.combiners.Count.PerKey())
# This reuses the existing cluster.
ib.collect(delays_by_airline)
# Describe the cluster running the pipelines.
# You can access the Flink Dashboard from the printed link.
ib.clusters.describe()
# Cleans up all long-lasting clusters managed by the notebook session.
ib.clusters.cleanup(force=True)
ノートブック管理のクラスタ
- デフォルトでは、パイプライン オプションが指定されていない場合、Interactive Beam は、
FlinkRunnerでパイプラインを実行するために、最後に使用されたクラスタを常に再利用します。- この動作を回避するには、たとえば、ノートブックでホストされていない FlinkRunner を使用して同じノートブック セッションで別のパイプラインを実行する場合は、
ib.clusters.set_default_cluster(None)を実行します。
- この動作を回避するには、たとえば、ノートブックでホストされていない FlinkRunner を使用して同じノートブック セッションで別のパイプラインを実行する場合は、
- 既存の Dataproc クラスタにマッピングされたプロジェクト、リージョン、プロビジョニング構成を使用する新しいパイプラインをインスタンス化する場合、Dataflow はクラスタを再利用します（これは直近に使用されたクラスタではない場合があります）。
- ただし、プロビジョニングの変更（クラスタのサイズ変更など）が行われるたびに、必要な変更を有効にするために新しいクラスタが作成されます。クラスタのサイズを変更する場合は、
ib.clusters.cleanup(pipeline)を使用して不要なクラスタをクリーンアップすることにより、クラウド リソースの枯渇を回避します。 - Flink
master_urlが指定されていて、これがノートブック セッションによって管理されているクラスタに属している場合、Dataflow はマネージド クラスタを再利用します。master_urlがノートブック セッションで不明な場合、ユーザーのセルフホスト型のFlinkRunnerが必要であることを意味します。ノートブックは暗黙的に何も実行しません。
注意事項とよくある質問
Flink IOException: ネットワーク バッファの数が不足している
わかりやすくするために、Flink ネットワーク バッファ構成は公開されません。
ジョブグラフが複雑すぎたり、並列処理の設定が高すぎると、ステップのカーディナリティ * 並列処理が大きくなりすぎるため、並行してスケジュールされるタスクが多くなり、実行に失敗する場合があります。
インタラクティブな実行速度を向上させるためのヒント:
- 検査する PCollection だけを変数に割り当てる
- PCollection を 1 つずつ検査します。
- 高ファンアウト変換後にリシャッフルを使用します。
- データサイズに基づいて並列処理を調整します（小さいほど高速になる場合があります）。
データの検査に時間がかかりすぎる
Flink ダッシュボードで、実行中のジョブを確認します。処理中のデータは単一のマシン上に存在し、シャッフルされないため、数百のタスクが完了して 1 つだけタスクが残っているステップが表示されることがあります。
次のような高度なファンアウト変換の後に必ず Reshuffle を使用してください。
- ファイルから行を読み取る
- BigQuery テーブルからの行の読み取り
リシャッフルを行わないと、ファンアウト データは常に同じタスク / ワーカーで実行され、並列処理はどれだけ高く設定しても利用できません。
ワーカーの数はどのくらい必要か
目安として、Flink クラスタには約 vCPU * #TMs 個のスロットがあります。40 個の n1-highmem-8 ワーカーがあるとします。Flink クラスタには最大で 8 × 40 = 320 スロットがあります。
READ->MAP->COMBINE ジョブを数百に並列処理して設定することにより、数千のタスクを同時に並列処理できるようになるはずです。
ストリーミングへの対応
ストリーミング パイプラインは、現時点では、ノートブック管理クラスタでのインタラクティブな Flink の機能と互換性がありません。
Beam SQL と beam_sql マジック
Beam SQL では、SQL ステートメントを使用して、制限付き PCollection と制限なし PCollection をクエリできます。Apache Beam ノートブックで作業している場合は、IPython カスタム マジック beam_sql を使用して、パイプラインの開発を高速化できます。
-h または --help オプションを使用して、beam_sql マジックの使用方法を確認できます。
定数値から PCollection を作成できます。
複数の PCollection を結合できます。
-r DataflowRunner または --runner DataflowRunner オプションを指定して Dataflow ジョブを起動できます。
詳細については、サンプル ノートブック Apache Beam SQL in notebooks をご覧ください。
JIT コンパイラと GPU を使用して高速化する
numba や GPU などのライブラリを使用して、Python コードと Apache Beam パイプラインを高速化できます。nvidia-tesla-t4 GPU で作成された Apache Beam ノートブック インスタンスで、numba.cuda.jit を使用して Python コードをコンパイルし、GPU で実行できます。必要に応じて、numba.jit または numba.njit を使用して Python コードをマシンコードにコンパイルし、CPU での実行を高速化することもできます。
GPU で処理される DoFn。
class Sampler(beam.DoFn):
def __init__(self, blocks=80, threads_per_block=64):
# Uses only 1 cuda grid with below config.
self.blocks = blocks
self.threads_per_block = threads_per_block
def setup(self):
import numpy as np
# An array on host as the prototype of arrays on GPU to
# hold accumulated sub count of points in the circle.
self.h_acc = np.zeros(
self.threads_per_block * self.blocks, dtype=np.float32)
def process(self, element: Tuple[int, int]):
from numba import cuda
from numba.cuda.random import create_xoroshiro128p_states
from numba.cuda.random import xoroshiro128p_uniform_float32
@cuda.jit
def gpu_monte_carlo_pi_sampler(rng_states, sub_sample_size, acc):
"""Uses GPU to sample random values and accumulates the sub count
of values within a circle of radius 1.
"""
pos = cuda.grid(1)
if pos < acc.shape[0]:
sub_acc = 0
for i in range(sub_sample_size):
x = xoroshiro128p_uniform_float32(rng_states, pos)
y = xoroshiro128p_uniform_float32(rng_states, pos)
if (x * x + y * y) <= 1.0:
sub_acc += 1
acc[pos] = sub_acc
rng_seed, sample_size = element
d_acc = cuda.to_device(self.h_acc)
sample_size_per_thread = sample_size // self.h_acc.shape[0]
rng_states = create_xoroshiro128p_states(self.h_acc.shape[0], seed=rng_seed)
gpu_monte_carlo_pi_sampler[self.blocks, self.threads_per_block](
rng_states, sample_size_per_thread, d_acc)
yield d_acc.copy_to_host()
GPU での実行:
詳細については、サンプル ノートブック Use GPUs with Apache Beam をご覧ください。
カスタム コンテナをビルドする
ほとんどの場合、パイプラインで追加の Python 依存関係や実行可能ファイルが不要な場合、Apache Beam で自動的に正式なコンテナ イメージを使用して、ユーザー定義コードを実行できます。これらのイメージには一般的な Python モジュールが数多く付属しているため、ビルドしたり、明示的に指定する必要はありません。
ただし、場合によっては、Python の追加の依存関係や Python 以外の依存関係が存在することがあります。このようなシナリオでは、カスタム コンテナをビルドして、Flink クラスタで実行できるようにします。カスタム コンテナを使用するメリットは次のとおりです。
- 連続的またはインタラクティブな実行の設定時間が短縮される
- 安定した構成 / 依存関係
- 柔軟性の向上: Python の以外の依存関係も設定できます。
コンテナのビルドプロセスは面倒なものになることがありますが、以下の使用パターンに従うことで、ノートブックですべての作業を行えるようになります。
ローカル ワークスペースを作成する
まず、Jupyter ホーム ディレクトリの下にローカル作業ディレクトリを作成します。
!mkdir -p /home/jupyter/.flink
Python 依存関係を準備する
さらに、使用する可能性のあるすべての追加の Python 依存関係をインストールして、要件ファイルにエクスポートします。
%pip install dep_a
%pip install dep_b
...
%%writefile ノートブック マジックを使用して、要件ファイルを明示的に作成できます。
%%writefile /home/jupyter/.flink/requirements.txt
dep_a
dep_b
...
または、すべてのローカル依存関係を要件ファイルに固定することもできます（これにより、意図しない依存関係が生じる可能性があります）。
%pip freeze > /home/jupyter/.flink/requirements.txt
Python 以外の依存関係を準備する
Python 以外のすべての依存関係をワークスペースにコピーします（ない場合は、この手順をスキップします）。
!cp /path/to/your-dep /home/jupyter/.flink/your-dep
...
Dockerfile を作成する
%%writefile ノートブック マジックを使用して、Dockerfile を作成します。例:
%%writefile /home/jupyter/.flink/Dockerfile
FROM apache/beam_python3.7_sdk:2.40.0
COPY requirements.txt /tmp/requirements.txt
COPY your_dep /tmp/your_dep
...
RUN python -m pip install -r /tmp/requirements.txt
このサンプル コンテナは、Python3.7 をベースとする Beam v2.40.0 のイメージを使用し、your_dep ファイルを追加して、追加の Python 依存関係をインストールします。この Dockerfile をテンプレートとして使用し、ユースケースに合わせて編集します。
Beam パイプラインで Python 以外の依存関係を参照する場合は、COPY の宛先を使用します。たとえば、/tmp/your_dep は your_dep ファイルのパスです。
Cloud Build でコンテナ イメージをビルドする
最後に、Cloud Build を使用してコンテナ イメージをビルドし、Container Registry に保存します。
# Enable the Cloud Build service if not enabled.
!gcloud services enable cloudbuild.googleapis.com
# Enable the Container Registry service if not enabled.
!gcloud services enable containerregistry.googleapis.com
!cd /home/jupyter/.flink \
&& gcloud builds submit \
--tag gcr.io/$(gcloud config get-value project)/flink:latest \
--timeout=20m
カスタム コンテナを使用する
カスタム コンテナの使用方法は、ランナーに応じて異なります。
Apache Beam コンテナの一般的な使用方法については、以下をご覧ください。
Dataflow コンテナの使用方法については、以下をご覧ください。