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. データセットのマウント：Notebook ノードの編集ページで、右側のナビゲーションウィンドウの [スケジューリング設定] > [スケジューリングポリシー] に移動します。データセットを追加します。

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 セッションを作成し、Livy を停止します。
   %maxcompute_spark

2. PySpark コードの実行：接続が確立された後、%%spark セルマジックを使用して、新しい Python セルで 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 ゲートウェイに接続することで、EMR Serverless Spark と連携できます。

1. 接続の確立：コマンドを実行する前に、セルの右下隅で接続する EMR コンピュートリソースと Livy ゲートウェイを選択します。

   # 基本的な接続
   %emr_serverless_spark
   
   # または、接続時にカスタム Spark パラメーターを渡します。カスタム Spark パラメーターを定義する際は、2 つのパーセント記号 (%%) を使用する必要があることに注意してください。
   %%emr_serverless_spark
   {
     "spark_conf": {
       "spark.emr.serverless.environmentId": "<EMR Serverless Spark ランタイム環境 ID>",
       "spark.emr.serverless.network.service.name": "<EMR Serverless Spark ネットワーク接続 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 ゲートウェイページでトークンを削除した場合、このコマンドを使用して再作成できます。

   # 現在の開発環境の 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 ゲートウェイを共有している場合、`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=<ADB の vSwitch ID> \
    --spark-conf spark.adb.eni.securityGroupId=<開発環境のセキュリティグループ ID>

   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