DataWorks:高度なノートブック開発

Python リソースの作成および公開

1. DataWorks DataStudio の左側ナビゲーションウィンドウで、 をクリックして Resource Management に移動します。

2. 対象ディレクトリを右クリックするか、右上隅の + アイコンをクリックします。[リソースの作成] > [MaxCompute Python] を選択し、ファイル名を my_utils.py とします。

3. [ファイルの内容] セクションで、[オンライン編集] をクリックし、ユーティリティ機能コードを貼り付けて、[保存] をクリックします。

   # my_utils.py
   def greet(name):
       return f"Hello, {name} from resource file!"

4. [保存] をクリックした後、ツールバーの [公開] をクリックします。これにより、リソースが開発および本番タスクの両方で利用可能になります。

ノートブックでのリソース参照

Python セルの先頭行で、##@resource_reference を使用して公開済みリソースを参照します。

##@resource_reference{"my_utils.py"}
# リソースがサブディレクトリにある場合 (例: my_folder/my_utils.py)、
# ディレクトリ名を除いて参照します: ##@resource_reference{"my_utils.py"}
from my_utils import greet

message = greet('DataWorks')
print(message)

開発環境でのデバッグ

Python セルを実行すると、次の出力が表示されます。

Hello, DataWorks from resource file!

本番環境への公開

ノートブックノードを保存および公開した後、[オペレーションセンター] > [定期実行タスク] に移動し、[テスト] をクリックします。タスクが成功すると、ログに Hello, DataWorks from resource file! という出力が表示されます。

詳細については、「MaxCompute リソースおよび関数」をご参照ください。

開発環境でのデバッグ

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"Data successfully written to 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"Data successfully written to MaxCompute table mc_test_table.")

詳細については、「個人用開発環境でのデータセットの使用」をご参照ください。

%odps — PyODPS エントリーオブジェクトの取得

現在の MaxCompute プロジェクトにバインドされた認証済み PyODPS オブジェクトを返します。これにより、コード内に AccessKey をハードコードする必要がなくなります。

o = %odps

コマンド実行後、右下隅にコンピュートリソースセレクターが表示され、プロジェクトが自動的に選択されます。プロジェクト名をクリックして別のプロジェクトに切り替えることができます。

このオブジェクトを使用して PyODPS スクリプトを実行します。たとえば、現在のプロジェクト内のすべてのテーブルを一覧表示するには、次のとおりです。

with o.execute_sql('show tables').open_reader() as reader:
    print(reader.raw)

%maxframe — MaxFrame 接続の確立

MaxCompute データ上で分散型かつ pandas 風のデータ処理を行うための MaxFrame セッションを作成します。

# MaxFrame を使用して MaxCompute に接続します。
mf_session = %maxframe

df = mf_session.read_odps_table('your_mc_table')
print(df.head())

# デバッグ後に手動でセッションを破棄してリソースを解放します。
mf_session.destroy()

エンジン比較

Spark エンジンに接続後、ノートブックカーネル全体の実行コンテキストがリモート PySpark 環境に切り替わります。以降のセルでは直接 PySpark コードを記述してください。

MaxCompute Spark

MaxCompute コンピュートリソースのバインドを接続前に完了させてください。

MaxCompute プロジェクトに組み込まれた Spark エンジンに Livy 経由で接続します。

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

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

2. PySpark コードの実行：新しい Python セルで %%spark セルマジックを使用します。

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

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

   # Spark セッションおよび Livy を停止します。
   %maxcompute_spark stop
   
   # Livy を停止して削除します。
   %maxcompute_spark delete

EMR Serverless Spark

EMR Serverless Spark コンピュートリソースのバインドおよび Livy Gateway の作成を接続前に完了させてください。

事前に作成した Livy Gateway に接続して EMR Serverless Spark と連携します。

1. 接続の確立：セルの右下隅で EMR コンピュートリソースおよび Livy gateway を選択し、次のいずれかのコマンドを実行します。

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

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

   # 基本的な接続
   %emr_serverless_spark

   カスタム Spark パラメーターを渡すには、%%emr_serverless_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"
     }
   }

   カスタムパラメーターは現在のセッションにのみ適用されます。省略した場合、システムは Admin Center で設定されたグローバルパラメーターを使用します。複数のタスクおよびユーザー間で構成を共有するには、[Admin Center] > [Serverless Spark] > [SPARK parameters] でグローバルに設定します。[Prioritize Global Configurations] オプションは、両方に同じパラメーターが存在する場合の優先度を制御します。

2. （任意）再接続：管理者が Livy gateway ページでトークンを削除した場合は、次のコマンドで再作成します。

   # Livy トークンを再接続およびリフレッシュします。
   %emr_serverless_spark refresh_token

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

   • EMR Spark SQL セルを使用して EMR Serverless Spark に SQL を送信 — セルは %emr_serverless_spark から接続を再利用し、ジョブを自動的に送信します。セル内でコンピュートリソースを選択する必要はありません。

   • Python セルで PySpark コードを送信 — %%spark プレフィックスは不要です。

4. 接続の解放：

   重要

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

   # Spark セッションおよび Livy を停止します。
   %emr_serverless_spark stop
   
   # Livy を停止して削除します。
   %emr_serverless_spark delete

AnalyticDB for Spark

AnalyticDB for Spark コンピュートリソースのバインドを接続前に完了させてください。

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

1. 接続の確立：セルの右下隅で ADB Spark コンピュートリソース を選択します。ネットワーク接続を確保するために vSwitch ID および Security Group ID を設定し、次のコマンドを実行します。

   • vSwitch ID (`vswitchId`)：Alibaba Cloud AnalyticDB MySQL コンソールで、インスタンス詳細ページの [ネットワーク情報] 内の [vSwitch ID] を確認します。

   • Security Group ID (`securityGroupId`)：個人用開発環境の詳細ページの [ネットワーク設定] で、選択した [セキュリティグループ] の ID（sg- で始まる）を確認します。

   重要

   ネットワーク接続を確保するには、個人用開発環境を作成する際に AnalyticDB for Spark インスタンスと同じ Virtual Private Cloud (VPC) および vSwitch を選択してください。

   # ネットワーク接続のための vSwitch ID および Security Group 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 および Security Group ID の確認方法：

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

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

   # AnalyticDB for Spark は C-Store テーブルのみを処理できます。
   df = spark.sql("SELECT * FROM my_adb_cstore_table LIMIT 10")
   df.show()

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

   %adb_spark cleanup

接続の確立

Python セルで %lindorm_ray を実行します。右下隅にコンピュートリソースセレクターが表示されるので、Lindorm コンピュートリソースおよび作成済みの RAY リソースグループ を選択します。

# 指定された Lindorm Ray リソースグループに接続します。
%lindorm_ray

Ray コードの実行

接続成功後、新しい Python セルで Ray コードを記述および実行します。ログはリアルタイムでセルの出力エリアにストリームバックされます。

次の例では、@ray.remote デコレーターを使用してリモートタスクを定義します。このタスクは Ray クラスター上で実行され、そのログおよび結果が出力エリアに返されます。

import ray
import time

@ray.remote
def hello_world():
  print("Hello from Lindorm Ray!")
  time.sleep(5)
  return "Task finished."

# リモートタスクを送信します。
result_ref = hello_world.remote()
print(ray.get(result_ref))

カスタム起動パラメーターの指定（任意）

サードパーティ製 Python パッケージをインストールする場合やローカルコードファイルをアップロードする場合は、%%lindorm_ray を使用してカスタムランタイム環境構成とともに接続を確立します。

例 1：依存関係のインストール

pip パラメーターを使用して、Ray 環境内に jieba パッケージをインストールします。

%%lindorm_ray
{
  "runtime_env": {
    "pip": ["jieba"]
  }
}

環境準備後、後続の Ray タスクでパッケージをインポートおよび使用します。

import ray

@ray.remote
def do_work(x):
    import jieba
    return "/".join(jieba.cut(x))

print(ray.get(do_work.remote("Welcome to the DataWorks+LindormRay solution")))

例 2：DataWorks リソースのアップロードおよび使用

working_dir パラメーターを使用して、DataWorks リソース管理から Ray クラスターにリソースをアップロードします。

# DataStudio リソース管理からリソースを参照し、そのパスを宣言します。
%%lindorm_ray
{
    "runtime_env": {
        "working_dir": "/mnt/workspace/_dataworks/resource_references"
    }
}

ray_resource.py が DataStudio リソース管理にアップロード済みであると仮定します。

ray_resource.py:

def fun():
    print("This is a test function in ray_resource.py")

これを Ray タスクで参照および使用します。

import ray

##@resource_reference{"ray_resource.py"}

@ray.remote
def do_work(x):
    print('Ray says:', x)
    from ray_resource import fun
    fun()
    return x

worker = do_work.remote("Welcome to the DataWorks+LindormRay solution")
print(ray.get(worker))

本番スケジューリングおよび運用管理

開発およびデバッグ後、ノートブックノードをコミットおよび公開します。これにより、ノードは DAG 内で定期スケジュールに基づき Lindorm Ray ノードとして実行されます。

• パラメーター化：標準の DataWorks スケジューリングパラメーター（例： ${bizdate}）を使用します。

• ログの確認：過剰なログがパフォーマンスに影響を与えないようにするため、システムはデフォルトで最初の 1 MB のログのみを読み込みます。ログが切り捨てられた場合、完全なタスブログへのリンクが Lindorm コンソールに表示されます。

• リソース解放：スケジュールされた本番タスク終了後、Lindorm Ray タスクは望ましい状態に入り、リソースを解放します。インタラクティブ開発中は、カーネルを再起動するかノートブックを閉じてリソースを解放します。