PyODPS は、DataWorks などのデータ開発プラットフォームでデータ開発ノードとして呼び出すことができます。これらのプラットフォームは PyODPS の実行環境とスケジュール実行機能を提供するため、ODPS エントリを手動で作成する必要はありません。PyODPS は、Pandas に似た、高速で柔軟かつ表現力豊かなデータ構造をサポートしています。PyODPS が提供する DataFrame API を使用して、Pandas のデータ結果処理機能を利用できます。本稿では、DataWorks プラットフォームを例に、PyODPS をすぐに使い始め、実際のプロジェクトで使用できるようになるまでの手順を説明します。
前提条件
-
MaxCompute の有効化が完了していること。
-
MaxCompute プロジェクトの作成が完了していること。
操作手順
-
PyODPS ノードの新規作成
本稿では、DataWorks の PyODPS ノードを使用して開発します。詳細については、「Develop a PyODPS 3 job」をご参照ください。
-
PyODPS 3 ノードを例として使用します。PyODPS 3 ノードの基盤となる Python のバージョンは 3.7 です。
-
PyODPS ノードがローカル処理のために取得するデータ量は 50 MB、ノード実行時に使用されるメモリは 1 GB を超えることはできません。超えた場合、ノードタスクはシステムによって中止されます。したがって、PyODPS タスクに大量のデータを処理する Python コードを記述することは避けてください。
-
DataWorks 上でコードを記述してデバッグするのは効率的ではないため、開発効率を向上させるには、ローカルに IDEA をインストールしてコード開発を行うことを推奨します。
-
ビジネスフローの新規作成
データ開発 ページに移動し、Business Flow を右クリックして、Create Workflow を選択します。
-
PyODPS ノードの新規作成
新しく作成したビジネスフローを右クリックし、 を選択し、ノード名を入力して 提出 をクリックします。
-
-
PyODPS ノードの編集
-
プログラムコードの記述
PyODPS ノードの編集ボックスにテストコードを入力します。以下は、PyODPS API を使用してテーブル操作を実行する完全な例です。テーブル操作と SQL 操作の詳細については、「テーブル」および「SQL」をご参照ください。
from odps import ODPS # フィールド名とフィールドタイプを直接指定して、非パーティションテーブル my_new_table を作成します。 # DataWorks の PyODPS ノードには、デフォルトでグローバル変数 odps または o が含まれています。これらは ODPS エントリです。ODPS エントリを手動で定義する必要はなく、そのまま使用できます。詳細は、「DataWorks での PyODPS の使用」をご参照ください。 table = o.create_table('my_new_table', 'num bigint, id string', if_not_exists=True) # 非パーティションテーブル my_new_table にデータを挿入します。 records = [[111, 'aaa'], [222, 'bbb'], [333, 'ccc'], [444, '中文']] o.write_table(table, records) # 非パーティションテーブル my_new_table からデータを読み取ります。 for record in o.read_table(table): print(record[0],record[1]) # SQL を実行してテーブルからデータを読み取ります。 result = o.execute_sql('select * from my_new_table;',hints={'odps.sql.allow.fullscan': 'true'}) # SQL の実行結果を読み取ります。 with result.open_reader() as reader: for record in reader: print(record[0],record[1]) # リソースを解放するためにテーブルを削除します。 table.drop() -
コードの実行
編集が完了したら、
アイコン をクリックします。実行が終了すると、下部の Runtime Log に実行結果が表示されます。次のログが出力されると、実行は成功です。2023-07-21 15:06:41 INFO ======================================================== Executing user script with PyODPS 0.11.2.3 Tunnel session created: <TableUploadSession xxx > Tunnel session created: <TableDownloadSession xxx > 111 aaa 222 bbb 333 ccc 444 中文 Tunnel session created: <InstanceDownloadSession id=xxx project_name=xxx > 111 aaa 222 bbb 333 ccc 444 中文 2023-07-21 15:06:49 INFO ======================================================== 2023-07-21 15:06:49 INFO Exit code of the Shell command 0 2023-07-21 15:06:49 INFO --- Invocation of Shell command completed --- 2023-07-21 15:06:49 INFO Shell run successfully! 2023-07-21 15:06:49 INFO Current task status: FINISH 2023-07-21 15:06:49 INFO Cost time is: 7.507s /home/admin/alisatasknode/taskinfo/xxx xxx .log-END-EOF
-