Lightning Engine で Spark ワークロードを高速化する

このドキュメントでは、Lightning Engine を有効にして、Managed Service for Apache Spark バッチ ワークロードとインタラクティブ セッションを高速化する方法について説明します。

概要

Lightning Engine は、マルチレイヤ最適化エンジンを搭載した高性能クエリ アクセラレータです。このエンジンは、クエリや実行の最適化などの従来の最適化手法と、ファイル システム レイヤやデータアクセス コネクタでのキュレートされた最適化の両方を行います。

次の図に示すように、Lightning Engine は、Spark クエリ の実行パフォーマンスをTPC-Hのようなワークロード (10 TB のデータセット サイズ) で高速化します。

注: パフォーマンスの結果は、10 TB の Cloud Storage Parquet データ とTPC-H 標準から派生したクエリに基づいています。これらの結果は、標準仕様のすべての要件を満たしていないため、公開されている TPC-H 標準の結果と比較することはできません。

詳細については、 Lightning Engine のご紹介 - Apache Spark を次世代のパフォーマンスにをご覧ください。

Lightning Engine の可用性

• Lightning Engine は、 Managed Service for Apache Spark ランタイム 2.3 で使用できます。
• Lightning Engine は、Managed Service for Apache Spark プレミアム料金階層でのみ使用できます。
  • バッチ ワークロード: Lightning Engine は、プレミアム階層のバッチ ワークロードで自動的に有効になります。必要なご対応は特にありません。
  • インタラクティブ セッション: Lightning Engine は、インタラクティブ セッションではデフォルトで有効になっていません。有効にするには、 Lightning Engine を有効にするをご覧ください。
  • セッション テンプレート: Lightning Engine は、セッション テンプレートではデフォルトで有効になっていません。有効にするには、 Lightning Engine を有効にするをご覧ください。

Lightning Engine を有効にする

次のセクションでは、Managed Service for Apache Spark バッチ ワークロード、セッション テンプレート、インタラクティブ セッションで Lightning Engine を有効にする方法について説明します。

バッチ ワークロード

バッチ ワークロードで Lightning Engine を有効にする

コンソール、Google Cloud CLI、Dataproc API を使用して、バッチ ワークロードで Lightning Engine を有効にできます。 Google Cloud

コンソール

コンソールを使用して、バッチ ワークロードで Lightning Engine を有効にします。 Google Cloud

1. コンソールで次の操作を行います。 Google Cloud

   1. Managed Service for Apache Spark バッチに移動します。
   2. [作成] をクリックして [バッチ作成] ページを開きます。

2. 次のフィールドを選択して入力します。

   • コンテナ:
     • ランタイム バージョン: 2.3 を選択します。

   • 階層構成:

     • [Premium] を選択します。これにより、[LIGHTNING ENGINE を有効にして Spark のパフォーマンスを向上させる] が自動的に有効になり、オンになります。

     プレミアム階層を選択すると、ドライバ コンピューティング階層 とエグゼキュータ コンピューティング階層 が Premium に設定されます。 3.0 より前のランタイムを使用するバッチでは、これらの自動的に設定されるプレミアム ティアのコンピューティング設定をオーバーライドすることはできません。

     ドライバ ディスク階層 とエグゼキュータ ディスク階層 を Premium に構成するか、デフォルトの Standard 階層の値のままにすることができます。プレミアム ディスク階層を選択する場合は、ディスクサイズを選択する必要があります。詳細については、 リソース割り当てプロパティをご覧ください。

   • プロパティ: 省略可: Key（プロパティ名）と Value のペアを入力して、 ネイティブ クエリ実行ランタイムを選択します:

     キー	値
     spark.dataproc.lightningEngine.runtime	先住民

     ネイティブ クエリ実行を有効にするために以前使用されていた spark.dataproc.runtimeEngine=native プロパティは非推奨になりました。3.0+ ランタイム以降ではサポートされていません。

3. 他のバッチ ワークロードの設定を入力、選択、確認します。 Spark バッチ ワークロードを送信するをご覧ください。

4. [送信] をクリックして Spark バッチ ワークロードを実行します。

gcloud

次の gcloud CLI gcloud dataproc batches submit spark コマンド フラグを設定して、バッチ ワークロードで Lightning Engine を有効にします。

gcloud dataproc batches submit spark \
    --project=PROJECT_ID \
    --region=REGION \
    --version=2.3 \
    --properties=dataproc.tier=premium \
    OTHER_FLAGS_AS_NEEDED

注:

• PROJECT_ID: 実際の Google Cloud プロジェクト ID。 プロジェクト ID は、 Google Cloud コンソールのダッシュボードの [プロジェクト情報] セクションに表示されます。
• REGION: ワークロードを実行できる利用可能な Compute Engine リージョン。

• --properties=dataproc.tier=premium。プレミアム階層を設定すると、バッチ ワークロードに次のプロパティが自動的に設定されます。

  • spark.dataproc.engine=lightningEngine は、バッチ ワークロードに Lightning Engine を選択します。
  • spark.dataproc.driver.compute.tier と spark.dataproc.executor.compute.tier は premium に設定されます（リソース割り当てプロパティを参照）。3.0 より前のランタイムを使用する Batch では、プレミアム ティアのコンピューティング設定を自動的にオーバーライドすることはできません。

• その他のプロパティ

  • ネイティブ クエリエンジン: spark.dataproc.lightningEngine.runtime=native ネイティブ クエリ実行ランタイムを選択する場合は、このプロパティを追加します。

    ネイティブ クエリ実行を有効にするために以前使用されていた spark.dataproc.runtimeEngine=native プロパティは非推奨になりました。3.0+ ランタイム以降ではサポートされていません。

  • ディスク階層とサイズ: デフォルトでは、ドライバとエグゼキュータ ディスクサイズは standard 階層とサイズに設定されます。プロパティを追加して、premium ディスク階層とサイズ（375 GiB の倍数）を選択できます。
    詳細については、リソース割り当てプロパティをご覧ください。

• OTHER_FLAGS_AS_NEEDED: Spark バッチ ワークロードを送信するをご覧ください。

API

バッチ ワークロードで Lightning Engine を有効にするには、 batches.create リクエストの一部として次の操作を行います。

• RuntimeConfig.version を 2.3 に設定します。

• RuntimeConfig.properties

  • spark.dataproc.engine=lightningEngine は、バッチ ワークロードに Lightning Engine を選択します。
  • spark.dataproc.driver.compute.tier と spark.dataproc.executor.compute.tier は premium に設定されます（リソース割り当てプロパティを参照）。3.0 より前のランタイムを使用するバッチでは、これらの自動的に設定されたプレミアム ティアのコンピューティング設定をオーバーライドすることはできません。

その他の RuntimeConfig.properties:

• ネイティブ クエリエンジン: spark.dataproc.lightningEngine.runtime:native。 ネイティブ クエリ実行ランタイムを選択する場合は、このプロパティを追加します。

  ネイティブ クエリ実行を有効にするために以前使用されていた spark.dataproc.runtimeEngine=native プロパティは非推奨になりました。3.0+ ランタイム以降ではサポートされていません。

• ディスク階層とサイズ: デフォルトでは、ドライバとエグゼキュータ ディスクサイズは standard 階層とサイズに設定されます。プロパティを追加して、premium 階層とサイズ（375 GiB の倍数）を選択できます。
  詳細については、リソース割り当てプロパティをご覧ください。

他のバッチ ワークロード API フィールドを設定するには、Spark バッチ ワークロードを送信する をご覧ください。

セッション テンプレート

セッション テンプレートで Lightning Engine を有効にする

コンソール、Google Cloud CLI、Dataproc API を使用して、Jupyter またはSpark Connect セッションのセッション テンプレートで Lightning Engine を有効にできます。 Google Cloud

コンソール

コンソールを使用して、バッチ ワークロードで Lightning Engine を有効にします。 Google Cloud

1. コンソールで次の操作を行います。 Google Cloud

   1. Managed Service for Apache Spark セッション テンプレートに移動します。
   2. [作成] をクリックして [セッション テンプレートの作成] ページを開きます。

2. 次のフィールドを選択して入力します。

   • セッション テンプレート情報:
     • [Lightning Engine を有効にして Spark のパフォーマンスを向上させる] を選択します。
   • 実行構成:
     • ランタイム バージョン: 2.3 を選択します。

   • プロパティ: 次の Key（プロパティ名）と Value のペアを入力して、 プレミアム階層を選択します:

     キー	値
     dataproc.tier	プレミアム
     spark.dataproc.engine	lightningEngine

     省略可: ネイティブ クエリ実行ランタイムを選択する場合は、次の Key（プロパティ名）とValueのペアを入力します。

     キー	値
     spark.dataproc.lightningEngine.runtime	native

     ネイティブ クエリ実行を有効にするために以前使用されていた spark.dataproc.runtimeEngine=native プロパティは非推奨になりました。3.0+ ランタイム以降ではサポートされていません。

3. 他のセッション テンプレートの設定を入力、選択、確認します。セッション テンプレートを作成する をご覧ください。

4. [送信] をクリックしてセッション テンプレートを作成します。

gcloud

gcloud CLI を使用して、Managed Service for Apache Spark セッション テンプレートを直接作成することはできません。代わりに、 gcloud beta dataproc session-templates import コマンドを使用して既存のセッション テンプレートをインポートし、インポートしたテンプレートを編集して Lightning Engine とネイティブ クエリ ランタイム（省略可）を有効にしてから、 編集したテンプレートを gcloud beta dataproc session-templates export コマンドを使用してエクスポートします。

API

セッション テンプレートで Lightning Engine を有効にするには、 sessionTemplates.create リクエストの一部として次の操作を行います。

• RuntimeConfig.version を 2.3 に設定します。
• RuntimeConfig.properties

その他の RuntimeConfig.properties:

• ネイティブ クエリエンジン: spark.dataproc.lightningEngine.runtime:native: RuntimeConfig.properties にこのプロパティを追加して、ネイティブ クエリ実行 ランタイムを選択します。
  ネイティブ クエリ実行を有効にするために以前使用されていた spark.dataproc.runtimeEngine=native プロパティは非推奨になりました。3.0+ ランタイム以降ではサポートされていません。

他のセッション テンプレート API フィールドを設定するには、セッション テンプレートを作成する をご覧ください。

インタラクティブ セッション

インタラクティブ セッションで Lightning Engine を有効にする

Google Cloud CLI または Dataproc API を使用して、Managed Service for Apache Spark インタラクティブ セッションで Lightning Engine を有効にできます。 BigQuery Studio ノートブックのインタラクティブ セッションで Lightning Engine を有効にすることもできます。

gcloud

次の gcloud CLI gcloud beta dataproc sessions create spark コマンド フラグを設定して、インタラクティブ セッションで Lightning Engine を有効にします。

gcloud beta dataproc sessions create spark \
    --project=PROJECT_ID \
    --location=REGION \
    --version=2.3 \
    --properties=dataproc.tier=premium,spark.dataproc.engine=lightningEngine \
    OTHER_FLAGS_AS_NEEDED

注:

• PROJECT_ID: 実際の Google Cloud プロジェクト ID。 プロジェクト ID は、 Google Cloud コンソールのダッシュボードの [プロジェクト情報] セクションに表示されます。
• REGION: ワークロードを実行できる利用可能な Compute Engine リージョン。

• --properties=dataproc.tier=premium,spark.dataproc.engine=lightningEngine。これらのプロパティにより、セッションで Lightning Engine が有効になります。

• その他のプロパティ:

  • ネイティブ クエリエンジン: spark.dataproc.lightningEngine.runtime=native: このプロパティを追加して、 ネイティブ クエリ実行ランタイムを選択します。
    ネイティブ クエリ実行を有効にするために以前使用されていた spark.dataproc.runtimeEngine=native プロパティは非推奨になりました。3.0+ ランタイム以降ではサポートされていません。

• OTHER_FLAGS_AS_NEEDED: インタラクティブ セッションを作成するをご覧ください。

API

セッションで Lightning Engine を有効にするには、 sessions.create リクエストの一部として次の操作を行います。

• RuntimeConfig.version を 2.3 に設定します。
• RuntimeConfig.properties

その他の RuntimeConfig.properties:

• ネイティブ クエリエンジン: spark.dataproc.lightningEngine.runtime:native: ネイティブ クエリ実行ランタイムを選択する場合は、このプロパティをRuntimeConfig.propertiesに追加します。
  ネイティブ クエリ実行を有効にするために以前使用されていた spark.dataproc.runtimeEngine=native プロパティは非推奨になりました。3.0+ ランタイム以降ではサポートされていません。

他のセッション テンプレート API フィールドを設定するには、インタラクティブ セッションを作成する をご覧ください。

BigQuery ノートブック

BigQuery Studio PySpark ノートブックでセッションを作成するときに、Lightning Engine を有効にできます。

from google.cloud.dataproc_spark_connect import DataprocSparkSession
from google.cloud.dataproc_v1 import Session
session = Session()

# Enable Lightning Engine.
session.runtime_config.properties["version"] = "2.3"
session.runtime_config.properties["dataproc.tier"] = "premium"
session.runtime_config.properties["spark.dataproc.engine"] = "lightningEngine"

# Enable THE Native Query Execution runtime.
session.runtime_config.properties["spark.dataproc.lightningEngine.runtime"] = "native"

# Create the Spark session.
spark = (
   DataprocSparkSession.builder
     .appName("APP_NAME")
     .dataprocSessionConfig(session)
     .getOrCreate())

# Add Spark application code here:

Lightning Engine の設定を確認する

コンソール Google Cloud 、Google Cloud CLI、Dataproc API を使用して、バッチ ワークロード、セッション テンプレート、 インタラクティブ セッションで Lightning Engine の設定を確認できます。

バッチ ワークロード

• バッチの階層 が premium に設定され、エンジン が Lightning Engine に設定されていることを確認するには:

  • Google Cloud コンソール: [バッチ] ページで、 バッチの [**階層**] 列と [**エンジン**] 列を確認します。[バッチ ID] をクリックして、バッチの詳細ページでこれらの設定を表示することもできます。
  • gcloud CLI: gcloud dataproc batches describe コマンドを実行します。
  • API: batches.get リクエストを発行します。

セッション テンプレート

• セッション テンプレートのエンジン が Lightning Engine に設定されていることを確認するには:

  • Google Cloud コンソール: [セッション テンプレート] ページで、 テンプレートの [**エンジン**] 列を確認します。セッション テンプレートの [名前] をクリックして、セッション テンプレートの詳細ページでこの設定を表示することもできます。
  • gcloud CLI: gcloud beta dataproc session-templates describe コマンドを実行します。
  • API: sessionTemplates.get リクエストを発行します。

インタラクティブ セッション

• インタラクティブ セッションのエンジン が Lightning Engine に設定されていることを確認するには:

  • Google Cloud コンソール: [インタラクティブ セッション] ページで、 テンプレートの [**エンジン**] 列を確認します。[インタラクティブ セッション ID] をクリックして、セッション テンプレートの詳細ページでこの設定を表示することもできます。
  • gcloud CLI: gcloud beta dataproc sessions describe コマンドを実行します。
  • API: sessions.get リクエストを発行します。

ネイティブ クエリ実行

ネイティブ クエリ実行（NQE）は、Google ハードウェア向けに設計された Apache Gluten と Velox に基づくネイティブな実装によってパフォーマンスを向上させる、Lightning Engine のオプション機能です。

ネイティブ クエリ実行ランタイムには統合メモリ管理が含まれており、既存の Spark 構成を変更することなく、オフヒープ メモリとオンヒープ メモリを動的に切り替えられます。NQE は、演算子、関数、Spark データ型への対応を拡張しているほか、ネイティブ エンジンを活用してオペレーションのプッシュダウンを最適化する機会を自動的に特定するインテリジェンスも備えています。

ネイティブ クエリ実行ワークロードを特定する

次のシナリオでは、ネイティブ クエリ実行を使用します。

• Parquet ファイルと ORC ファイルからデータを読み取る Spark Dataframe API、Spark Dataset API、Spark SQL クエリ。出力ファイルの形式は、ネイティブ クエリ実行のパフォーマンスに影響しません。

• ネイティブ クエリ実行の評価ツールで推奨されるワークロード。

ネイティブ クエリ実行は、次のデータ型の入力を使用するワークロードではおすすめしません。

• バイト: ORC と Parquet
• タイムスタンプ: ORC
• 構造体、配列、マップ: Parquet

ネイティブ クエリ実行の制限事項

次のシナリオでネイティブ クエリ実行を有効にすると、例外、Spark の非互換性、ワークロードのデフォルトの Spark エンジンへのフォールバックが発生する可能性があります。

フォールバック

次の実行でネイティブ クエリ実行を行うと、ワークロードが Spark 実行エンジンにフォールバックし、回帰または失敗が発生する可能性があります。

• ANSI: ANSI モードが有効になっている場合、実行は Spark にフォールバックします。

• 大文字と小文字を区別するモード: ネイティブ クエリ実行は、Spark のデフォルトの大文字と小文字を区別しないモードのみをサポートしています。大文字と小文字を区別するモードが有効になっている場合、誤った結果が生じる可能性があります。

• パーティション分割テーブル スキャン: ネイティブ クエリ実行は、パスにパーティション情報が含まれている場合にのみ、パーティション分割テーブル スキャンをサポートします。それ以外の場合、ワークロードは Spark 実行エンジンにフォールバックします。

互換性のない動作

次の場合にネイティブ クエリ実行を使用すると、互換性のない動作や誤った結果が生じる可能性があります。

• JSON 関数: ネイティブ クエリ実行は、単一引用符ではなく二重引用符で囲まれた文字列をサポートします。単一引用符を使用すると、誤った結果が生じます。get_json_object 関数でパスに「*」を使用すると、NULL が返されます。

• Parquet 読み取り構成:

  • ネイティブ クエリ実行では、 spark.files.ignoreCorruptFiles が true に設定されている場合でも、デフォルトの false 値に設定されているとみなされます。
  • ネイティブ クエリ実行は spark.sql.parquet.datetimeRebaseModeInRead, を無視し、Parquet ファイルの内容のみを返します。従来のハイブリッド（ユリウス暦とグレゴリオ暦）と先発グレゴリオ暦の違いは考慮されません。Spark の結果が異なる場合があります。

• NaN: 対象外です。たとえば、数値比較で NaN を使用すると、予期しない結果が生じる可能性があります。

• Spark カラム型読み取り: Spark カラム型ベクトルはネイティブ クエリ実行と互換性がないため、致命的なエラーが発生する可能性があります。

• スピル: シャッフル パーティションが大きな数に設定されている場合、ディスクへのスピル機能により OutOfMemoryException がトリガーされる可能性があります。この問題が発生した場合は、パーティションの数を減らすことでこの例外を解消できます。

注: ネイティブ クエリ実行の評価ツールを実行して、Managed Service for Apache Spark バッチ ワークロードとインタラクティブ セッションでネイティブ クエリ実行ランタイムを使用するかどうかを判断できます。