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

プロジェクトリソース (.py ファイル) の参照

コードをよりモジュール化し、再利用性および保守性を高めるために、共通の関数やクラスを独立した .py ファイルにまとめます。その後、##@resource_reference{"custom_name.py"} 構文を使用して、これらのファイルを MaxCompute リソースとして参照できます。

1. Python リソースの作成と公開

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

   2. Resource Management のディレクトリツリーで、対象ディレクトリを右クリックするか、右上隅の + をクリックします。New Resource > [MaxCompute Python] を選択し、ファイル名を my_utils.py とします。

   3. Document Content セクションで、Online Editing をクリックし、ユーティリティ関数のコードをコードエディタに貼り付け、Save をクリックします。

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

   4. ツールバーで Save をクリックし、次に Publish をクリックしてリソースを公開します。これにより、リソースが開発環境および本番環境の両方のタスクで利用可能になります。

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

   ノートブックの 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)

3. 開発環境でのデバッグ

   Python セルを実行します。出力結果は次のとおりです。

   Hello, DataWorks from resource file!

   重要

   開発環境でのデバッグ中に、システムは ##@resource_reference 宣言を検出し、Resource Management から対応するファイルを個人ディレクトリ内の workspace/_dataworks/resource_references パスに自動的にダウンロードし、コードからアクセスできるようにします。

   ModuleNotFoundError が発生した場合は、エディタツールバーの 再起動 ボタンをクリックしてリソースを再読み込みし、再度試してください。

4. 本番環境への公開と検証

   Save および Publish をクリックしてノートブックノードを公開した後、Operation and Maintenance Center > Auto Triggered Task に移動し、[テスト] をクリックしてタスクを実行します。タスクが成功すると、ログに Hello, DataWorks from resource file! が出力されます。

   重要

   There is no file with id ... エラーが発生した場合は、Python リソースが本番環境に公開されていることを確認してください。

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

データセットの読み書き (OSS/NAS)

ノートブックタスクの実行中、OSS または NAS に格納された大規模ファイルを簡単に読み書きできます。

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

ワークスペースパラメーターの使用

DataWorks では、既存のスケジューリングパラメーターを拡張するワークスペースパラメーターを提供しています。これらのパラメーターにより、タスクおよびノード間でグローバル設定を再利用し、環境を隔離できます。SQL セルまたは Python セルで、${workspace.param} 形式を使用してワークスペースパラメーターを参照できます。ここで、param はワークスペースパラメーターの名前です。

1. ワークスペースパラメーターの作成：開始前に、DataWorks で Operation and Maintenance Center > Scheduling Settings > Workspace Parameters に移動して、必要なパラメーターを作成します。

2. ワークスペースパラメーターの参照：

• SQL セルでワークスペースパラメーターを参照します。

  SELECT '${workspace.param}';

  セルを実行すると、ワークスペースパラメーターの解決済み値が出力されます。

• Python セルでワークスペースパラメーターを参照します。

  print('${workspace.param}')

  セルを実行すると、ワークスペースパラメーターの解決済み値が出力されます。

詳細については、「ワークスペースパラメーターの使用」をご参照ください。

MaxCompute への接続

• %odps：PyODPS エントリオブジェクトの取得

  このコマンドは、現在の MaxCompute プロジェクトにバインドされた認証済みの PyODPS オブジェクトを返します。AccessKey をコード内にハードコードしないため、MaxCompute との連携にはこの方法を推奨します。

  1. マジックコマンドを使用して MaxCompute 接続を作成します。%odps を入力します。右下隅に MaxCompute コンピュートリソースセレクターが表示され、自動的にコンピュートリソースが選択されます。右下隅の MaxCompute プロジェクト名をクリックしてプロジェクトを切り替えられます。

     o=%odps 

  2. 取得した MaxCompute コンピュートリソースを使用して、PyODPS スクリプトを実行します。

     たとえば、現在のプロジェクト内のすべてのテーブルを取得するには、次のとおりです。

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

• %maxframe：MaxFrame 接続の確立

  このコマンドは MaxFrame セッションを作成し、MaxCompute 向けの分散型 pandas ライクなデータ処理機能を提供します。

  # MaxCompute MaxFrame セッションへの接続とアクセス
  mf_session = %maxframe
  
  df = mf_session.read_odps_table('your_mc_table')
  print(df.head())
  
  # 開発およびデバッグ後は、手動でセッションを破棄してリソースを解放します
  mf_session.destroy()

Spark コンピュートリソースへの接続

DataWorks Notebook は複数の Spark エンジンへの接続をサポートしています。これらのエンジンは、接続方法、実行コンテキスト、リソース管理の点で異なります。

Lindorm Ray への接続

Lindorm コンピュートエンジンの RAY リソースグループは分散コンピューティングサービスを提供し、エンドツーエンドの AI ワークロードをサポートします。マジックコマンドを使用して、ノートブックから Lindorm Ray リソースにシームレスに接続し、インタラクティブな開発およびデバッグを行い、その後ノートブックをスケジュールされた本番タスクとして公開できます。

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

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

   重要

   • Lindorm Ray コンピュートリソースに接続した後、同じノートブックで SQL セルを実行できなくなります。Lindorm Ray エンジンは Python および Ray コードのみを排他的に実行します。

   • 同じコードセルを複数回実行すると、システムが自動的に前の Ray ジョブを終了し、新しいジョブを開始します。これにより、リソースの無駄遣いやタスクの競合を防げます。

2. Ray コードの実行：接続確立後、新しい Python セルに直接 Ray コードを記述および実行できます。ログはリアルタイムでセルの出力エリアにストリーミングされ、インタラクティブなデバッグが容易になります。

   次の例では、Ray クラスター上で実行され、ログおよび最終結果をセルの出力エリアに返すシンプルなリモートタスク (@ray.remote デコレーターを使用) を定義します。

   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))

3. (オプション) カスタム起動パラメーターの指定：Ray 環境にサードパーティ製 Python パッケージをインストールしたり、ローカルのコードファイルをアップロードしたりするなど、追加の構成を指定する必要がある場合は、%%lindorm_ray コマンドを使用して接続を確立します。

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

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

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

     環境の準備が完了したら、後続の Ray ジョブでパッケージをインポートおよび使用できます。次の例は、リモート関数内で jieba を呼び出して中国語の形態素解析を実行する方法を示しています。

     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 Resource Management から Ray クラスターにリソースをアップロードし、タスク内でインポートおよび呼び出せるようにするために使用します。

     重要

     • working_dir を使用してリソースをアップロードする場合、ファイルは開発環境から直接 Ray クラスターにアップロードされ、100 MB のサイズ制限が適用されます。リソースパッケージが大きすぎると、アップロードが失敗したり、Ray ノードが不安定になったりする可能性があります。

     • 大規模なファイルまたは依存関係 (>100 MB) の場合は、OSS にアップロードしてコードからアクセスするか、カスタムイメージにパッケージ化してください。このアプローチにより、安定性とパフォーマンスが向上します。

     # DataStudio Resource Management にアップロードされたリソースを参照し、そのパスを宣言します。
     %%lindorm_ray
     {
         "runtime_env": {
             "working_dir": "/mnt/workspace/_dataworks/resource_references"
         }
     }

     DataStudio の Resource Management に ray_resource.py ファイルをアップロードしたと仮定します。次のセルを記述および実行すると、システムが後続のコード内の ##@resource_reference 宣言を自動的に解析し、対応するリソースを /mnt/workspace/_dataworks/resource_references パスにダウンロードします。

     ray_resource.py のサンプルコード：

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

     重要

     開発環境では、##@resource_reference を含むセルを実行した後、上記の %%lindorm_ray セルを再度実行して、working_dir 内のダウンロード済みリソースを 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))

4. 本番スケジューリングおよび運用：開発およびデバッグ完了後、ノートブックノードをコミットおよび公開できます。その後、DAG 内の Lindorm Ray ノードとして定期的にスケジュールされます。

   • パラメーター化：コードでは、${bizdate} などの標準的な DataWorks スケジューリングパラメーターを使用できます。

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

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

付録：マジックコマンド クイックリファレンス