MaxCompute は Graph 専用の開発プラグインを提供していません。Eclipse を使用して MaxCompute Graph プログラムを開発できます。
開発ワークフローは次のとおりです：
- Graph コードを記述し、ローカルデバッグを使用して基本的なテストを実行します。
- クラスターデバッグを実行して結果を検証します。
開発例
このセクションでは、単一始点最短経路 (SSSP) アルゴリズムを例に、Eclipse で Graph プログラムを開発およびデバッグする方法を説明します。
手順：
- graph_examples という名前の Java プロジェクトを作成します。
- MaxCompute クライアントの lib ディレクトリから JAR パッケージを Eclipse プロジェクトの [Java ビルド・パス] に追加します。[Java ビルド・パス] ページの [ライブラリ] タブで、mapreduce-api.jar を選択して展開し、[Javadoc ロケーション] をダブルクリックします。[Javadoc URL] を選択し、
http://odps.alibaba-inc.com/doc/prdoc/odps_graph/api/を入力します。 - MaxCompute Graph プログラムを開発します。
一般的な方法は、単一始点最短経路アルゴリズムなどの既存の例をコピーして変更することです。この例では、パッケージパスのみを package com.aliyun.odps.graph.example に変更します。
- プログラムをコンパイルしてパッケージ化します。
Eclipse で、ソースコードディレクトリ (図の src ディレクトリ) を右クリックし、 を選択して JAR パッケージを生成します。JAR パッケージの宛先パス (例：D:\\odps\\clt\\odps-graph-example-sssp.jar) を選択します。
- MaxCompute クライアントを使用して SSSP ジョブを実行します。詳細については、「Graph ジョブの実行」をご参照ください。
ローカルデバッグ
MaxCompute Graph は、Eclipse でブレークポイントデバッグを実行できるローカルデバッグモードをサポートしています。
手順：
- odps-graph-local Maven パッケージをダウンロードします。
- Eclipse プロジェクトで、Graph ジョブのメインプログラムファイル (
main関数を含むファイル) を右クリックし、 を選択します。 - [引数] タブで、[プログラムの引数] を 1 sssp_in sssp_out に設定します。
- [引数] タブで、[VM 引数] を次のように設定します。
また、[プログラムの引数] でプログラムパラメーターを設定します。たとえば、-Dodps.runner.mode=local -Dodps.project.name=<project.name> -Dodps.end.point=<end.point> -Dodps.access.id=<access.id> -Dodps.access.key=<access.key>1 sssp_in sssp_outを使用します。これらはそれぞれ、開始ノード、入力テーブル名、出力テーブル名を表します。 - odps.end.point パラメーターが指定されていないローカルモードの場合、warehouse ディレクトリに sssp_in テーブルと sssp_out テーブルを作成し、入力テーブル sssp_in にデータを追加する必要があります。次のコードは、サンプル入力データを示しています。
1,"2:2,3:1,4:4" 2,"1:2,3:2,4:1" 3,"1:1,2:2,5:1" 4,"1:4,2:1,5:1" 5,"3:1,4:1"warehouse ディレクトリの詳細については、「ローカルでのジョブの実行」をご参照ください。
- [実行] をクリックして、SSSP ジョブをローカルで実行します。説明 パラメーター設定については、MaxCompute クライアントの conf/odps_config.ini ファイルをご参照ください。上記のパラメーターは一般的に使用されるものです。次のリストでこれらのパラメーターについて説明します：
- odps.runner.mode：値は local に設定する必要があります。このパラメーターはローカルデバッグに必須です。
- odps.project.name：現在のプロジェクトを指定します。このパラメーターは必須です。
- odps.end.point：MaxCompute サービスのエンドポイントを指定します。このパラメーターはオプションです。このパラメーターを省略すると、テーブルとリソースはローカルの warehouse ディレクトリから読み取られます。データまたはメタデータが存在しない場合は、例外がスローされます。このパラメーターを指定すると、システムはまずローカルの warehouse からの読み取りを試みます。要求されたデータまたはメタデータが存在しない場合、システムは MaxCompute からの読み取りにフォールバックします。
- odps.access.id：MaxCompute サービスに接続するための AccessKey ID です。このパラメーターは、odps.end.point が指定されている場合にのみ有効です。
- odps.access.key：MaxCompute サービスに接続するための AccessKey Secret です。このパラメーターは、odps.end.point が指定されている場合にのみ有効です。
- odps.cache.resources：使用するリソースのリストを指定します。これは、JAR コマンドの
-resourcesオプションに相当します。 - odps.local.warehouse：ローカルの warehouse ディレクトリへのパスです。このパラメーターが指定されていない場合、デフォルトのパスは ./warehouse です。
以下は、Eclipse で SSSP ジョブをローカルで実行したときのデバッグ出力です。Counters: 3 com.aliyun.odps.graph.local.COUNTER TASK_INPUT_BYTE=211 TASK_INPUT_RECORD=5 TASK_OUTPUT_BYTE=161 TASK_OUTPUT_RECORD=5 graph task finish説明 上記の例では、ローカルの warehouse ディレクトリに sssp_in テーブルと sssp_out テーブルが含まれている必要があります。sssp_in テーブルと sssp_out テーブルの詳細については、「Graph プログラムの作成」をご参照ください。
ローカルジョブ用の一時ディレクトリ
ローカルデバッグセッションを実行するたびに、Eclipse プロジェクトディレクトリに一時ディレクトリが作成されます。一時ディレクトリ名の例は
graph_20130816154834_240_5772 です。これには、次のサブディレクトリとファイルが含まれています：counters (カウンター情報)、inputs (zhemin_test1.sssp_in などの入力データ)、outputs (_default_ サブディレクトリを含む出力結果)、resources (centers サブディレクトリを含むリソースファイル)、superSteps (スーパーステップ情報)、および job.xml ジョブ設定ファイル。ローカルで実行される Graph ジョブの一時ディレクトリには、次のディレクトリとファイルが含まれます：
- counters：ジョブの実行中に生成されたカウンター情報が含まれます。
- inputs：ジョブの入力データが含まれます。システムはまず、ローカルの warehouse からデータを取得しようとします。データが見つからず、odps.end.point パラメーターが設定されている場合、システムは MaxCompute SDK を使用してサーバーからデータを読み取ります。デフォルトでは、各 input に対して最大 10 レコードが読み取られます。この制限は
-Dodps.mapred.local.record.limitパラメーターを使用して変更できますが、値は 10,000 を超えることはできません。 - outputswarehouseoutputs：ジョブの出力データが含まれます。ジョブが完了すると、このディレクトリの結果データがローカルの warehouse 内の対応するテーブルを上書きします。
- resources：ジョブで使用されるリソースが含まれます。入力と同様に、システムはまずローカルの warehouse からリソースを取得しようとします。リソースが見つからない場合、odps.end.point パラメーターが設定されていれば、システムは MaxCompute SDK を使用してサーバーからリソースを読み取ります。
- job.xml：ジョブの設定が含まれます。
- superstep：各反復の永続化情報を格納します。
説明 ローカルデバッグ中に詳細なログを出力するには、log4j.properties_odps_graph_local_debug という名前の log4j 設定ファイルを src ディレクトリに配置します。
クラスターデバッグ
ローカルデバッグが完了したら、ジョブをクラスターに送信してテストできます。
手順：
- MaxCompute クライアントを設定します。
add jar /path/work.jar -f;コマンドを使用して JAR パッケージを更新します。- JAR コマンドを使用してジョブを実行し、実行ログと結果データを確認します。
説明 クラスターで Graph ジョブを実行する方法の詳細については、「Graph プログラムの作成」をご参照ください。
パフォーマンスの最適化
次の Graph ジョブ設定項目はパフォーマンスに影響します：
setSplitSize(long)：入力テーブルのスプリットサイズ。単位：MB。値は 0 より大きい必要があります。デフォルト値：64。setNumWorkers(int)：ジョブのワーカー数。有効な値：[1, 1000]。デフォルト値：1。最適なワーカー数は、ジョブの入力サイズ (バイト単位) とsplitSizeによって異なります。setWorkerCPU(int)：各ワーカーの CPU リソース。値 100 は 1 CPU コアを表します。有効な値：[50, 800]。デフォルト値：200。setWorkerMemory(int)：各ワーカーのメモリリソース。単位：MB。有効な値：[256, 12288]。デフォルト値：4096。setMaxIteration(int)：最大反復回数。デフォルト値：-1。0 以下の値は、最大反復回数がジョブの終了条件ではないことを示します。setJobPriority(int)：ジョブの優先度。有効な値：[0, 9]。デフォルト値：9。値が大きいほど優先度が低くなります。
最適化に関する一般的な推奨事項：
setNumWorkersメソッドを使用してワーカー数を増やすことを検討してください。setSplitSizeメソッドを使用してスプリットサイズを小さくし、データロードを高速化することを検討してください。- 各ワーカーの CPU またはメモリリソースを増やします。
- 最大反復回数を設定します。高い精度を必要としないアプリケーションの場合は、反復回数を減らしてジョブを早期に終了させます。
setNumWorkers API と setSplitSize API を一緒に使用して、データロード速度を向上させることができます。setNumWorkers が workerNum に、setSplitSize が splitSize に設定され、合計入力サイズ (バイト単位) が inputSize であると仮定します。入力スプリット数は splitNum=inputSize/splitSize として計算されます。workerNum と splitNum の関係は次のとおりです：- シナリオ 1：
splitNumがworkerNumと等しい場合、各ワーカーは 1 つのスプリットのロードを担当します。 - シナリオ 2：
splitNumがworkerNumより大きい場合、各ワーカーは 1 つ以上のスプリットのロードを担当します。 - シナリオ 3：
splitNumがworkerNumより小さい場合、各ワーカーは 0 または 1 つのスプリットのロードを担当します。
したがって、データロードを高速化するには、最初の 2 つのシナリオのいずれかが満たされるように workerNum と splitSize を調整する必要があります。反復フェーズでは、workerNum を調整するだけで済みます。runtime partitioning を False に設定した場合は、setSplitSize を使用してワーカー数を制御するか、最初の 2 つのシナリオのいずれかが満たされるようにします。3 番目のシナリオが発生した場合、一部のワーカーにはスプリットが割り当てられません。これを防ぐには、JAR コマンドの前に set odps.graph.split.size=<m>; set odps.graph.worker.num=<n>; コマンドを使用します。これは、setNumWorkers と setSplitSize を使用するのと同じです。
もう 1 つの一般的なパフォーマンスの問題はデータスキューです。カウンターの出力では、一部のワーカーが他のワーカーよりも著しく多くの頂点またはエッジを処理しているように見えます。データスキューは通常、少数のキーが不釣り合いに多数の頂点、エッジ、またはメッセージに対応する場合に発生します。これらのキーは少数のワーカーに割り当てられるため、それらのワーカーのランタイムが長くなります。この問題を解決するには、次のいずれかの方法を使用します：
- コンバイナを使用してこれらのキーのメッセージをローカルで集約し、送信されるメッセージの数を減らします。
開発者はコンバイナを定義して、メッセージストレージのメモリ使用量を削減し、ネットワークトラフィックを減らすことで、ジョブの実行時間を短縮できます。
- ビジネスロジックを最適化します。大規模なデータセットを扱う場合、ディスクからのデータ読み取りが処理時間のかなりの部分を占めることがあります。読み取る必要のあるデータ量を減らすことで、全体的なスループットとジョブのパフォーマンスを向上させることができます。次の改善を行うことができます：
- 入力データ量を減らす：特定の意思決定アプリケーションでは、データのサンプリングされたサブセットを処理しても、結果の全体的な精度ではなく、その精度にのみ影響する場合があります。このような場合は、入力テーブルにインポートする前にデータをサンプリングすることを検討してください。
- 不要なフィールドの読み取りを避ける：MaxCompute Graph フレームワークの TableInfo クラスを使用すると、テーブル全体やパーティションではなく、特定の列 (列名の配列として渡される) を読み取ることができます。これにより、入力データ量も削減され、ジョブのパフォーマンスが向上します。
組み込み JAR パッケージ
Graph プログラムを実行すると、次の JAR パッケージがデフォルトで JVM にロードされます。これらのリソースをアップロードしたり、コマンドの
-libjars オプションに含めたりする必要はありません。- commons-codec-1.3.jar
- commons-io-2.0.1.jar
- commons-lang-2.5.jar
- commons-logging-1.0.4.jar
- commons-logging-api-1.0.4.jar
- guava-14.0.jar
- json.jar
- log4j-1.2.15.jar
- slf4j-api-1.4.3.jar
- slf4j-log4j12-1.4.3.jar
- xmlenc-0.52.jar
説明 JVM クラスパスでは、これらの組み込み JAR パッケージがユーザーの JAR パッケージよりも優先されるため、バージョン競合が発生する可能性があります。たとえば、プログラムが commons-codec-1.5.jar のクラスの関数を使用しているが、その関数が commons-codec-1.3.jar に存在しない場合があります。バージョン 1.3 の機能が不十分な場合は、MaxCompute が新しいバージョンにアップグレードされるのを待つ必要があります。