MaxCompute:DataWorks での PyODPS の使用

DataWorks の各 PyODPS ノードには、MaxCompute エントリポイントであるグローバル変数 odps または o が含まれています。MaxCompute エントリポイントを指定する必要はありません。ステートメントの例:

# pyodps_iris テーブルが存在するかどうかを確認します。
print(o.exist_table('pyodps_iris'))

True が返された場合、pyodps_iris テーブルは存在します。

• PyODPS ノードで SQL 文を実行します。たとえば、execute_sql()/run_sql() を使用して SQL 文を実行できます。データ定義言語（DDL）およびデータ操作言語（DML）の SQL 文のみを実行できます。

  説明

  MaxCompute クライアントで実行できる特定の SQL 文は、MaxCompute エントリオブジェクトの execute_sql() メソッドと run_sql() メソッドを使用して実行できない場合があります。非 DDL 文または非 DML 文を実行するには、他のメソッドを使用する必要があります。たとえば、GRANT 文または REVOKE 文を実行するには run_security_query メソッドを使用し、API 操作を呼び出すには run_xflow メソッドまたは execute_xflow メソッドを使用する必要があります。

• PyODPS ノードから SQL 実行結果を読み取ります。たとえば、open_reader() メソッドを使用して SQL 実行結果を読み取ることができます。

PyODPS ノードでの SQL 関連操作の詳細については、「SQL」をご参照ください。

歴史的な互換性の理由から、DataWorks ではデフォルトで インスタンストンネル は有効になっていません。この場合、Result インターフェイスが呼び出されて instance.open_reader が実行され、最大 10,000 件のデータレコードを読み取ることができ、複雑なデータ型のサポートにも問題があります。プロジェクトで データ保護メカニズムが有効になっておらず、すべてのデータを反復的にフェッチする必要がある場合、または Arrays などの複雑なデータ型フィールドを読み取る必要がある場合は、インスタンストンネル を有効にし、limit を無効にする必要があります。

• コードの実行中に limit を無効にします。

  次の文を実行して、インスタンストンネルをグローバルに有効にし、limit を削除できます。

  options.tunnel.use_instance_tunnel = True
  options.tunnel.limit_instance_tunnel = False # 読み取ることができるデータレコード数の制限を削除します。
  
  with instance.open_reader() as reader:
  # インスタンストンネルを使用してすべてのデータを読み取ります。
  # reader.count を使用してデータレコード数を取得できます。

• 現在のリーダーに対してのみ limit を無効にします。

  tunnel=True を open_reader() メソッドに追加して、現在の InstanceTunnel を open_reader() メソッドで有効にすることもできます。limit=False を open_reader() メソッドに追加して、現在の open_reader() メソッドで読み取ることができるデータレコード数の limit を削除することもできます。

  with instance.open_reader(tunnel=True, limit=False) as reader:
  # 現在の open_reader() メソッドはインスタンストンネルを使用して呼び出され、すべてのデータを読み取ることができます。

インスタンストンネルとデータ読み取りの制限の詳細については、「SQL 文の実行結果を取得する」をご参照ください。

DataWorks で PyODPS ノードを使用してコードを開発する場合、スケジューリングパラメーターも使用できます。たとえば、スケジューリングパラメーターを使用して、ノードのデータタイムスタンプを取得できます。DataWorks の PyODPS ノードと SQL ノードは、同じ方法でスケジューリングパラメーターを指定します。ただし、スケジューリングパラメーターは、PyODPS ノードと SQL ノードのコードでは異なる方法で参照されます。

• ${param_name} などの文字列は、SQL ノードのコードで使用されます。

• PyODPS ノードのコードが実行される前に、args という名前の辞書がグローバル変数に追加されます。これにより、コードへの悪影響を防ぎます。コードは、コードで ${param_name} をスケジューリングパラメーターに置き換える代わりに、args[param_name] メソッドを使用してスケジューリングパラメーターの値を取得します。

たとえば、DataWorks の PyODPS ノードの [スケジューリング構成] タブで、ds=${yyyymmdd}[基本プロパティ] セクションの [パラメーター] フィールドにスケジューリングパラメーター を指定できます。次に、ノードのコードで次のコマンドを指定して、パラメーター値を取得できます。

• ds パラメーターの値を取得します。

  print('ds=' + args['ds'])
  # ds パラメーターで指定された時刻を取得します（例: ds=20161116）。

• ds=${yyyymmdd} で指定されたパーティションのデータを取得します。

  o.get_table('table_name').get_partition('ds=' + args['ds'])
  # table_name で指定されたテーブルの ds で指定されたパーティションからデータを取得します。

hints パラメーターを使用して、ランタイムパラメーターを構成できます。hints パラメーターの値は DICT 型です。

o.execute_sql('select * from pyodps_iris', hints={'odps.sql.mapper.split.size': 16})

sql.settings パラメーターをグローバルに構成できます。関連するランタイムパラメーターは、実行ごとに自動的に追加されます。

from odps import options
options.sql.settings = {'odps.sql.mapper.split.size': 16}
o.execute_sql('select * from pyodps_iris') # hints パラメーターはグローバル設定に基づいて自動的に構成されます。

次の表に、DataWorks ノードにプリインストールされているサードパーティパッケージと、ノードのパッケージのバージョンを示します。

使用したいサードパーティパッケージが上記の表に含まれていない場合は、DataWorks が提供する load_resource_package メソッドを使用して、MaxCompute リソースからパッケージをダウンロードできます。pyodps-pack を使用してパッケージを生成した後、load_resource_package メソッドを使用してサードパーティパッケージを読み込むことができます。その後、パッケージの内容を DataWorks にインポートできます。pyodps-pack の使用方法の詳細については、「PyODPS 用のサードパーティパッケージを生成する」および「PyODPS ノードでサードパーティパッケージを参照する」をご参照ください。

例:

1. 次のコマンドを実行して、IP アドレスをパッケージ化します。

   pyodps-pack -o ipaddress-bundle.tar.gz ipaddress

2. ipaddress-bundle.tar.gz パッケージをリソースとしてアップロードして送信した後、次の方法で PyODPS 3 ノードのパッケージを使用できます。

   load_resource_package("ipaddress-bundle.tar.gz")
   import ipaddress

ダウンロードしたパッケージの合計サイズは 100 MB を超えることはできません。プリインストールされたパッケージのパッケージ化操作をスキップする場合は、パッケージ化中に pyodps-pack が提供する --exclude パラメーターを使用できます。たとえば、次のパッケージ化方法は、DataWorks 環境に存在する NumPy パッケージと pandas パッケージを除外します。

pyodps-pack -o bundle.tar.gz --exclude numpy --exclude pandas <YOUR_PACKAGE>