DataWorks:Notebook の高度な開発

開発環境でのデバッグ

1. データセットのマウント：ご利用の個人開発環境の詳細ページに移動し、[ストレージ構成] > [データセット] でデータセットを構成します。

2. コードでのデータアクセス：データセットは個人開発環境内のパスにマウントされます。コード内でこのパスを直接読み書きできます。

   # 個人開発環境でマウントパス /mnt/data/dataset でデータセットが構成されていると仮定します。
   import pandas as pd
   
   # マウントパスを直接使用します。
   file_path = '/mnt/data/dataset/testfile.csv'
   df = pd.read_csv(file_path)
   
   # PyODPS を使用して MaxCompute にデータを書き込みます。
   o = %odps
   o.write_table('mc_test_table', df, overwrite=True)
   print(f"Successfully wrote data to the MaxCompute table mc_test_table.")

本番デプロイメント

1. データセットのマウント：ノートブックエディターの右側ナビゲーションウィンドウで、[スケジューリング構成] > [スケジューリングポリシー] に移動し、データセットを追加します。

2. コードでのデータアクセス：ノートブックをコミットおよび公開すると、データセットは本番環境内のパスにマウントされます。コード内でこのパスを直接読み書きできます。

   # マウントパス /mnt/data/dataset でデータセットが構成されていると仮定します。
   import pandas as pd
   
   # マウントパスを直接使用します。
   file_path = '/mnt/data/dataset/testfile.csv'
   df = pd.read_csv(file_path)
   
   # PyODPS を使用して MaxCompute にデータを書き込みます。
   o = %odps
   o.write_table('mc_test_table', df, overwrite=True)
   print(f"Successfully wrote data to the MaxCompute table mc_test_table.")

エンジンの比較

MaxCompute Spark

Livy を介して、MaxCompute プロジェクトに組み込まれた Spark エンジンに接続します。

1. 接続の確立：Python セルで次のコマンドを実行します。システムは自動的に Spark セッションを作成または再利用します。

   # Spark セッションを作成します。
   %maxcompute_spark

2. PySpark コードの実行：接続が成功した後、新しい Python セル で %%spark セル Magic を使用して PySpark コードを実行します。

   # MaxCompute Spark を使用する場合、Python セルは %%spark で始める必要があります。
   %%spark
   
   df = spark.sql("SELECT * FROM your_mc_table LIMIT 10")
   df.show()

3. 接続の手動解放：開発およびデバッグ後、セッションを手動で停止または削除できます。本番環境では、システムが現在のタスクインスタンスの Livy セッションを自動的に停止および削除します。

   # Spark セッションをクリーンアップし、Livy を停止します。
   %maxcompute_spark stop
   
   # Spark セッションをクリーンアップし、Livy を停止してから削除します。
   %maxcompute_spark delete

EMR Serverless Spark

Livy Gateway に接続して、EMR Serverless Spark とやり取りします。

1. 接続の確立：コマンドを実行する前に、セルの右下隅で EMR 計算リソース および Livy Gateway を選択します。

   # 基本的な接続
   %emr_serverless_spark
   
   # または、接続時にカスタム Spark パラメーターを渡すこともできます。カスタム Spark パラメーターを指定する場合は、2 つのパーセント記号（%%）を使用する必要があることに注意してください。
   %%emr_serverless_spark
   {
     "spark_conf": {
       "spark.emr.serverless.environmentId": "<EMR_Serverless_Spark_runtime_environment_ID>",
       "spark.emr.serverless.network.service.name": "<EMR_Serverless_Spark_network_connection_ID>",
       "spark.driver.cores": "1",
       "spark.driver.memory": "8g",
       "spark.executor.cores": "1",
       "spark.executor.memory": "2g",
       "spark.driver.maxResultSize": "32g"
     }
   }

   説明

   カスタムパラメーターとグローバル設定の関係

   • デフォルトの動作：ここで定義されたカスタムパラメーターは、現在の接続（セッション）にのみ適用され、1 回限りで使用されます。カスタムパラメーターを指定しない場合、システムは自動的に [管理センター] で構成されたグローバルパラメーターを使用します。

   • 推奨される使用方法：複数のタスクまたは複数のユーザー間で再利用が必要な構成については、[管理センター] > [Serverless Spark] > [SPARK パラメーター] でグローバルパラメーターとして設定することを推奨します。これにより、一貫性が確保され、管理が簡素化されます。

   • 優先順位ルール：カスタムパラメーターとグローバル構成の両方に同じパラメーターが設定されている場合、[管理センター] の [グローバル構成を優先] オプションが、どちらのパラメーターが使用されるかを決定します。

     • 選択済み：グローバル構成がこのセッションのカスタムパラメーターをオーバーライドします。

     • 未選択：このセッションのカスタムパラメーターがグローバル構成をオーバーライドします。

2. （オプション）再接続：管理者が誤って Livy Gateway ページのトークンを削除した場合、このコマンドを使用して再作成できます。

   # 現在の個人開発環境の Livy トークンを再接続およびリフレッシュします。
   %emr_serverless_spark refresh_token

3. PySpark または SQL コードの実行：接続が成功すると、カーネルが切り替わります。Python セル に直接 PySpark コードを記述するか、EMR Spark SQL セル に SQL を記述できます。

   1. EMR Spark SQL セルによる SQL コードの送信および実行

      %emr_serverless_spark を使用して接続を確立した後、EMR Spark SQL セルに SQL 文を直接記述できます。セル内で計算リソースを選択する必要はありません。

      EMR Spark SQL セルは、%emr_serverless_spark からの接続を再利用し、ジョブを対象の計算リソースに送信します。

      

   2. Python セルによる PySpark コードの送信および実行

      %emr_serverless_spark を使用して接続を確立した後、新しい Python セルに PySpark コードを送信および実行できます。セルに %%spark プレフィックスを追加する必要はありません。

      

4. 接続の手動解放

   重要

   複数のユーザーが単一の Livy Gateway を共有している場合、stop または delete コマンドは、そのゲートウェイを使用中のすべてのユーザーに影響します。これらのコマンドは慎重に使用してください。

   # Spark セッションをクリーンアップし、Livy を停止します。
   %emr_serverless_spark stop
   
   # Spark セッションをクリーンアップし、Livy を停止してから削除します。
   %emr_serverless_spark delete

AnalyticDB for Spark

Spark Connect Server を作成して、AnalyticDB for Spark エンジンに接続します。

1. 接続の確立：ネットワーク接続を確保するため、接続パラメーターで vSwitch ID およびセキュリティグループ ID を構成する必要があります。コマンドを実行する前に、セルの右下隅で ADB Spark 計算リソース を選択します。

   # ネットワーク接続を確立するには、vSwitch ID およびセキュリティグループ ID を構成する必要があります。
   %adb_spark add \
    --spark-conf spark.adb.version=3.5 \
    --spark-conf spark.adb.eni.enabled=true \
    --spark-conf spark.adb.eni.vswitchId=<vSwitch_ID_of_ADB> \
    --spark-conf spark.adb.eni.securityGroupId=<Security_Group_ID_of_personal_development_environment>

   vSwitch ID およびセキュリティグループ ID の確認方法

   • vSwitch ID（vswitchId）：Alibaba Cloud [AnalyticDB for MySQL] コンソールに移動します。インスタンス詳細ページで、[vSwitch ID] を [ネットワーク情報] の下に表示します。

   • セキュリティグループ ID（securityGroupId）：ご利用の個人開発環境の詳細ページに移動します。[ネットワーク設定] の下で選択されている [セキュリティグループ] を見つけ、sg- で始まる ID を確認します。

     重要

     ネットワーク接続を確保するため、個人開発環境を作成する際に、AnalyticDB for Spark インスタンスと同じ VPC および vSwitch を選択することを推奨します。

2. PySpark コードの実行：接続が成功した後、新しい Python セル で PySpark コードを実行します。

   # C-Store テーブルでのみ操作を実行できます。
   df = spark.sql("SELECT * FROM my_adb_cstore_table LIMIT 10")
   df.show()

   注記：AnalyticDB for Spark エンジンは、'storagePolicy'='COLD' 属性を持つ C-Store テーブルのみを処理できます。

3. 接続の手動解放：開発環境でのデバッグ後、リソースを節約するために接続セッションを手動でクリーンアップします。本番環境では、システムがリソースを自動的にクリーンアップします。

   %adb_spark cleanup