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

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

共通の関数やクラスを .py ファイルにカプセル化し、##@resource_reference{"custom_name.py"} を使用して MaxCompute リソースとして参照します。これにより、コードがモジュール化され、メンテナンスが簡素化されます。

1. Python リソースの作成と発行

   1. DataWorks の [DataStudio] の左側メニューで、 アイコンをクリックして [リソース管理] に移動します。

   2. 「リソース管理」ツリーで、ターゲットディレクトリを右クリックするか、右上隅の [+] をクリックし、リソースの新規作成 > [MaxCompute Python] を選択して、リソースに my_utils.py という名前を付けます。

   3. ファイルの内容 セクションで オンライン編集 をクリックし、デバッグ済みのユーティリティ関数コードをエディターに貼り付けて 保存 をクリックします。

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

   4. ツールバーで 保存 をクリックし、次に 公開 をクリックします。これにより、リソースは 開発タスクと本番タスクの両方から参照できるようになります。

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 宣言を検出し、リソース管理から対象ファイルを個人のディレクトリ内の workspace/_dataworks/resource_references にダウンロードします。

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

4. 本番環境への発行と検証

   このノートブックノードを 保存 して 公開 した後、オペレーションセンター > 定期タスク に移動し、[Test] をクリックして実行します。タスクが正常に実行されると、ログに 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 で、オペレーションセンター > スケジュール設定 > ワークスペースパラメーター に移動してパラメーターを作成します。

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

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

  SELECT '${workspace.param}';

  クエリが正常に実行されると、パラメーターの解決済みの値が出力されます。

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

  print('${workspace.param}')

  コードが正常に実行されると、パラメーターの解決済みの値が出力されます。

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

MaxCompute への接続

• %odps：PyODPS エントリオブジェクトを取得します

  これにより、コード内で AccessKey をハードコーディングせずに済みます。

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

     o = %odps 

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

     たとえば、現在のプロジェクトにあるすべてのテーブルを取得するには、次のようにします。

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

• %maxframe: MaxFrame 接続を確立します

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

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

3. (オプション) カスタム起動パラメーターの指定: サードパーティの Python パッケージのインストールやローカルのコードファイルのアップロードなど、Ray 環境の追加設定を構成するには、%%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("欢迎使用DataWorks+LindormRay解决方案")))

   • 例 2：DataWorks リソースのアップロードと利用

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

     重要

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

     • 大容量のリソースや依存関係 (100 MB 超) の場合は、OSS にアップロードし、コードからダウンロードするか、カスタムイメージにパッケージ化して、安定性とパフォーマンスを高めてください。

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

     DataWorks のリソース管理に 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("欢迎使用DataWorks+LindormRay解决方案")
     print(ray.get(worker))

4. 本番のスケジューリングと運用保守：開発とデバッグが完了したら、このノートブックノードをコミットして発行します。DAG 内の Lindorm Ray ノードとして定期的にスケジューリングされます。

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

   • ログの確認：本番環境では、パフォーマンス問題を防ぐために、システムはデフォルトでログの先頭 1 MB のみを読み込みます。ログが途中で切れている場合、出力に Lindorm コンソールでタスクログ全体を確認するためのリンクが表示されます。

   • リソースの解放：スケジュールされた本番タスクが終了すると、Lindorm Ray タスクは終了状態になり、リソースを解放します。対話的な開発中は、カーネルを再起動するかノートブックを閉じて、タスクを終了してください。

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