MaxCompute:LogView を使用した診断ケース

実行に失敗したジョブの診断

ジョブの実行に失敗した場合、LogView の [Result] タブでエラー情報を確認できます。失敗したジョブの LogView を開くと、[Result] タブが自動的に表示されます。

実行が遅いジョブの診断

コンパイル段階

• スケジューリング

  問題の説明：ジョブのサブステータスが Waiting for cluster resource になっています。ジョブはコンパイルを待機しています。

  原因：コンピューティングクラスターのリソースが不足しています。

  解決策：コンピューティングクラスターのステータスと、コンピューティングクラスターの必須リソースを確認します。サブスクリプションクラスターを使用している場合は、リソースをスケールアウトできます。

• 最適化

  問題の説明：ジョブのサブステータスが SQLTask is optimizing query になっています。オプティマイザーが実行計画を最適化しています。

  原因：実行計画が複雑です。オプティマイザーが実行計画を最適化するのに長時間を要します。

  解決策：オプティマイザーが最適化を完了するまで待ちます。ほとんどの場合、このプロセスは 10 分未満で完了します。

• 実行計画の生成

  問題の説明：ジョブのサブステータスが SQLTask is generating execution plan になっています。

  原因 1：過剰な数のパーティションのデータが読み取られています。

  解決策：SQL ステートメントを最適化してパーティションの数を減らします。たとえば、パーティションプルーニングを実行したり、データを読み取る必要のないパーティションを除外したり、大きなジョブを小さなジョブに分割したりできます。SQL ステートメントでパーティションプルーニングが有効かどうかを判断する方法と、パーティションプルーニングが失敗する一般的なシナリオの詳細については、「パーティションプルーニングが有効かどうかの確認」をご参照ください。

  原因 2：過剰なスモールファイルが生成されています。スモールファイルは、次のシナリオで生成されます：

  1. Tunnel コマンドを使用してデータをアップロードする際に、誤った操作が実行されました。たとえば、データレコードをアップロードするたびに新しい upload session が作成されます。詳細については、「Tunnel コマンドに関するよくある質問」をご参照ください。

  2. パーティションテーブルに対して INSERT INTO ステートメントを実行すると、partition ディレクトリに新しいファイルが生成されます。

  解決策

  1. TunnelBufferedWriter インターフェイスを使用して、より効率的な方法でデータをアップロードします。これにより、過剰なスモールファイルが生成されるのを防ぎます。

  2. スモールファイルを手動でマージします。詳細については、「スモールファイルのマージ」をご参照ください。

  説明

  スモールファイルの数が 10,000 を超える場合は、スモールファイルの自動マージを有効にできます。システムは毎日スモールファイルを自動的にマージします。ただし、特別なシナリオでシステムがスモールファイルのマージに失敗した場合は、手動でスモールファイルをマージする必要があります。

• クラスター間のデータレプリケーション

  問題の説明: Task rerun が SubStatusHistory タブに複数回表示され、[結果] タブに FAILED: ODPS-0110141:Data version exception が表示される場合、ジョブは失敗していません。代わりに、ジョブはクラスター間でデータをレプリケートしています。

  原因 1：プロジェクトのデータがクラスター間で移行されています。この場合、移行完了後の最初の 1〜2 日間は、クラスター間でデータをレプリケーションする多数のジョブが実行されます。

  解決策：クラスター間のデータレプリケーションが期待どおりに完了するまで待ちます。

  原因 2：プロジェクトのデータがクラスター間で移行されています。ただし、パーティションが期待どおりにフィルター処理されていません。その結果、特定のパーティションから古いデータが読み取られます。

  解決策：古いデータを含むパーティションを除外します。

実行段階

実行計画は LogView の [Job Details] タブに表示されます。実行計画は完了しておらず、ジョブは Running 状態です。ジョブが実行段階でスタックしたり、この段階で完了するまでに予想以上に長い時間がかかったりする場合、この問題はリソースの待機、データスキュー、非効率的な UDF 実行、データ肥大化などの理由で発生する可能性があります。このセクションでは、各ケースの特性と解決策について説明します。

• リソースの待機

  現象：インスタンスが Ready 状態であるか、一部のインスタンスが Running 状態である一方で、他のインスタンスが Ready 状態です。インスタンスが Ready 状態であっても [Debug] 履歴エントリがある場合は、リソースの待機ではなく、インスタンスの障害によってトリガーされたリトライを示している可能性があることに注意してください。Fuxi インスタンスリストでは、[SmartFilter] を使用してステータスでフィルターできます。[Status] 列には、各インスタンスの現在の状態が表示されます。

  解決策：

  1. キューイング状態が正常かどうかを判断します。LogView の Queue length を表示することで、ジョブのキュー内の位置を確認できます。[Queue length] の値は、LogView の左側にある [Basic info] パネルで確認できます。または、MaxCompute コンソールで対応するクォータグループのリソース使用量を確認します。MaxCompute コンソールで、左側のメニューから [クォータ管理] > [リソース使用量] を選択し、ターゲットのクォータグループの [CPU リソース] などのメトリックの傾向グラフを表示します。リソースの使用量がクォータに近づいているか、超えている場合は、クォータグループのリソースが不足しており、キューイングが予想されることを示します。ジョブのスケジューリング順序は、送信時間と優先度だけでなく、必要なメモリまたは CPU リソースが満たされるかどうかにも依存します。

  2. クォータグループを使用するジョブを表示します。

     優先度の低い大規模なジョブが送信されたり、複数の小規模なジョブが一度に送信されたりする場合があります。これらのジョブは大量のリソースを占有します。ジョブの所有者に連絡してジョブを終了させ、占有されているリソースを解放することができます。

  3. ジョブのクォータグループを別のプロジェクトのクォータグループに変更します。

  4. リソースをスケールアウトします。この解決策は、サブスクリプションリソースを使用するユーザーにのみ適しています。

• データスキュー

  現象：タスク内のほとんどのインスタンスは終了していますが、いくつかの「ロングテール」インスタンスの終了に非常に長い時間がかかっています。これらの遅いインスタンスは、他よりも多くのデータを処理している可能性があります。[SmartFilter] で、[Long-tails] の数を確認し、[Latency] 列でインスタンスのレイテンシーを比較できます。

  解決策：データスキューの一般的な原因と関連する最適化方法の詳細については、「データスキューのチューニング」をご参照ください。

• 非効率的な UDF 実行

  このトピックでは、UDF は、ユーザー定義スカラー関数 (UDF)、ユーザー定義集約関数 (UDAF)、ユーザー定義テーブル値関数 (UDTF)、ユーザー定義結合 (UDJ)、およびユーザー定義型 (UDT) を含む、さまざまなユーザー定義の拡張機能を指します。

  説明：タスクの実行効率が低く、タスクに UDF が含まれています。UDF 実行タイムアウトを示す次のエラーメッセージが表示されることがあります：Fuxi job failed - WorkerRestart errCode:252,errMsg:kInstanceMonitorTimeout, usually caused by bad udf performance。

  トラブルシューティング方法：タスクが失敗した場合、LogView の [Job details] タブで DAG を調べて、UDF が含まれているかどうかを判断できます。DAG では、R4_3 のような失敗したタスクが赤で表示されます。ノード上の fx ラベルは、["JAVA"] のように UDF 言語を示します。R4_3 をダブルクリックしてオペレータービューを開くと、上から下への演算子チェーンが表示され、タスクで使用されているすべての UDF 名が一覧表示されます。タスクの [StdOut] ログでは、UDF フレームワークが入力レコード数、出力レコード数、および処理時間を出力します。このデータを使用して、パフォーマンスの問題を特定できます。通常、Speed(records/s) は数十万から数百万の範囲です。数万に低下した場合、パフォーマンスの問題が発生している可能性があります。ログには、UDF 処理の概要と、[OutputCount]、[InnerTime]、[Speed(records/s)] などのメトリックを持つ各オペレーター (CursorId) の統計テーブルが表示されます。

  解決策：パフォーマンスの問題が発生した場合は、次の方法を使用して問題をトラブルシューティングし、パフォーマンスを最適化できます。

  1. UDF でエラーが発生したかどうかを確認します。

     特定の場合、パフォーマンスの問題は特定のデータ値によって引き起こされます。たとえば、特定の値が表示されると無限ループが発生します。MaxCompute Studio を使用すると、テーブルの特定のサンプルデータをダウンロードし、オンプレミスマシンでそのデータを使用してトラブルシューティングを行うことができます。詳細については、MaxCompute Studio 開発マニュアルの「Java UDF」および「Python UDF」をご参照ください。

  2. UDF の名前がビルトイン関数と同じかどうかを確認します。

     ビルトイン関数は、同じ名前の UDF によって上書きされる可能性があります。関数がビルトイン関数のように見える場合は、同じ名前の UDF がビルトイン関数を上書きできるかどうかを判断する必要があります。

  3. UDF をビルトイン関数に置き換えます。

     類似の機能を提供するビルトイン関数が存在する場合は、UDF を使用しないことを推奨します。ビルトイン関数は検証済みであり、より効率的な方法で使用されます。さらに、オプティマイザーはビルトイン関数に対してホワイトボックステストを実行するため、より多くの最適化を行うことができます。ビルトイン関数の使用方法の詳細については、「ビルトイン関数」をご参照ください。

  4. 特定の UDF を類似の機能を提供するビルトイン関数に置き換え、ビルトイン関数を使用して実装できない UDF のみ保持します。

  5. UDF の `evaluate` メソッドを最適化します。

     `evaluate` メソッドは繰り返し実行されるため、パラメーターに関連する必要な操作でのみ使用します。初期化関連の操作や繰り返しの計算は事前に行います。

  6. UDF の実行に必要な時間を推定します。

     オンプレミスマシンで 1 つのインスタンスが処理するデータ量をシミュレートして、UDF の実行に必要な時間をテストします。その後、UDF の実装を最適化します。デフォルトでは、UDF の実行に必要な最大時間は 30 分です。UDF は 30 分以内にデータを返すか、context.progress() を使用してハートビートを報告する必要があります。UDF の実行に必要な推定時間が 30 分を超える場合は、パラメーターを設定して UDF のタイムアウト期間を指定できます。

     デフォルト値：1800。単位：秒。有効値：1〜3600。
     -- UDF のタイムアウト期間を指定します。単位：秒。デフォルト値：600。
     -- UDF のタイムアウト期間は [0,3600] の範囲で手動で調整できます。

  7. メモリパラメーターを変更します。

     UDF の効率が低いのは、必ずしも計算量のためではありません。ストレージの複雑さも UDF の効率に影響を与える可能性があります。例：

     • UDF が大量のデータに対してメモリコンピューティングやソートを実行すると、メモリオーバーフローが発生します。

     • メモリ不足により、ガベージコレクション (GC) の頻度が高くなります。

     メモリパラメーターを変更して、上記の問題に一時的に対処できます。具体的な最適化は、ビジネス要件に基づいて実行する必要があります。例：

     set odps.sql.udf.jvm.memory=
     -- UDF の JVM ヒープに使用できる最大メモリサイズを指定します。デフォルト値：1024。単位：MB。
     -- odps.sql.udf.jvm.memory パラメーターの値は [256,12288] の範囲で変更できます。

     説明

     UDF を使用すると、パーティションプルーニングが無効になる場合があります。MaxCompute V2.0 から UdfProperty アノテーションがサポートされています。UDF を定義するときに、アノテーションを使用して、UDF が決定論的であることをコンパイラに指定できます。サンプルコード：

     @com.aliyun.odps.udf.annotation.UdfProperty(isDeterministic = true)
     public class AnnotatedUdf extends com.aliyun.odps.udf.UDF {
         public String evaluate(String x) {
             return x;
         }
     }

     UDF の SQL ステートメントを書き換える場合は、パーティションフィルタリングで UDF を使用できます。

     -- 書き換え前の SQL ステートメント
     SELECT * FROM t WHERE pt = udf('foo');  -- pt は t のパーティションキー列を示します。
     -- 書き換え後の SQL ステートメント
     SELECT * FROM t WHERE pt = (SELECT udf('foo')); --pt は t のパーティションキー列を示します。

• データ肥大化

  現象：タスクからの出力データ量が入力データ量よりもはるかに大きい。たとえば、1 GB のデータが処理されて 1 TB になると、単一のインスタンスが 1 TB のデータを処理する際のパフォーマンスが大幅に低下します。ジョブが完了した後、タスクの [I/O Records] で入出力データ量を確認できます。ジョブが Join 段階でスタックした場合、Running 状態のいくつかの Fuxi インスタンスの [StdOut] ログを確認できます。[SmartFilter] で、[Running] 状態のインスタンスをフィルターし、インスタンスの [StdOut] アイコンをクリックしてそのログを表示します。[StdOut] ログに Merge Join ログが継続的に表示される場合、単一のワーカーが継続的に Merge Join を実行していることを示します。Merge Join からの出力行数が 1433 億を超えると、深刻なデータ肥大化を示しており、JOIN 条件と Join キーが適切かどうかを確認する必要があります。ログは継続的に [Merge join cursor] レコードを出力し、[OutputRowCount] フィールドの値は増加し続けます。これを使用して、データ肥大化の程度を判断できます。

  解決策：

  • コード内の次の問題を確認します：JOIN 条件が正しいか、JOIN 条件がデカルト積として記述されているか、UDTF が正常か、過剰な出力データが生成されているか。

  • データ肥大化が集約によって引き起こされているかどうかを確認します。

    ほとんどのアグリゲーターは再帰的な集約を実行します。アグリゲーターがデータを集約するとき、アグリゲーターはまず中間結果をマージします。中間結果のデータ量は大きくなく、ほとんどのアグリゲーターの計算量は低いです。データ量が大きい場合でも、集約は迅速に完了できます。ほとんどの場合、集約中にデータ肥大化は発生しません。ただし、次のシナリオで集約を実行すると、データ肥大化が発生する可能性があります：

    • SELECT ステートメントで集約を有効にして、異なるディメンションで DISTINCT 操作を実行します。DISTINCT 操作が実行されるたびにデータが拡張されます。

    • GROUPING SETS または CUBE | ROLLUP ステートメントを使用すると、中間結果データのサイズが元のデータサイズと比較して数倍に増加する可能性があります。ただし、COLLECT_LIST や MEDIAN などの特定のステートメントを使用する場合は、すべての中間結果データを保持する必要があります。これにより、特定の問題が発生する可能性があります。

  • JOIN 操作によるデータ肥大化を防ぎます。

    たとえば、2 つのテーブルを結合したいとします。左側テーブルには大量の人口データが含まれています。しかし、MaxCompute インスタンスの高い並列性のため、処理効率は高いです。右テーブルは、各性別の考えられる悪い習慣などの情報を記録するディメンションテーブルです。ディメンションテーブルには 2 つの性別しか含まれていませんが、各性別に対応する数百の行が存在します。性別でテーブルを結合すると、左側テーブルのデータが数百倍に拡張される可能性があります。データ肥大化を防ぐには、テーブルを結合する前に、右テーブルのデータ行を 2 つのデータ行に集約します。

  • GROUPING SET ステートメントによってデータ肥大化が発生しているかどうかを確認します。GROUPING SET ステートメントが実行されると、データが拡張され、出力データはグループ数と比較して数倍に増加します。現在の実行計画は GROUPING SET ステートメントに適応したり、ダウンストリームタスクの並列処理の次数を変更したりできません。ダウンストリームタスクの並列処理の次数を手動で設定できます。サンプルステートメント：

    set odps.stage.reducer.num = xxx; 
    set odps.stage.joiner.num = xxx;

終了段階

• 複数段階でのサブクエリ実行

  ほとんどの場合、MaxCompute SQL のサブクエリは同じ Fuxi DAG にコンパイルされます。したがって、すべてのサブクエリとメインクエリは 1 つの Fuxi ジョブで完了します。ただし、特定の特別なサブクエリは、メインクエリの前に個別に実行する必要があります。サンプルコード：

  SELECT
       product,
      sum(price)
   FROM
       sales
   WHERE
       ds in (SELECT DISTINCT ds FROM t_ds_set)
   GROUP BY product;

  サブクエリ SELECT DISTINCT ds FROM t_ds_set が最初に実行されます。その結果は、メインクエリが読み取るパーティションの数を最適化するためのパーティションプルーニングに必要です。これら 2 つの実行は、別々の Fuxi ジョブです。LogView は、各 Fuxi ジョブを SQL_0_0_0_job_0 や SQL_0_0_1_job_1 のような別のタブに表示します。DAG は、複数の JOB ノード間の依存関係も示します。2 番目の タブ をクリックするだけで、job_1 の実行ステータスを確認できます。切り替えた後、job_1 の各タスクの現在のステータス ( Running や Waiting など) を確認できます。

• 過剰なスモールファイル

  過剰なスモールファイルは、主にストレージパフォーマンスとコンピューティングパフォーマンスに影響します。

  • ストレージ：過剰なスモールファイルは、Pangu ストレージのワークロードを増加させます。これはストレージ使用量に影響します。

  • コンピューティング：単一の大きなファイルに対する MaxCompute の処理効率は、複数のスモールファイルに対する処理効率よりも高いため、全体的な処理パフォーマンスが影響を受けます。したがって、SQL ジョブが完了すると、特定の条件が満たされた場合にスモールファイルをマージする操作が自動的にトリガーされます。これにより、システムが過剰なスモールファイルを生成するのを防ぎます。

• 解決策：LogView を使用して、ジョブが自動スモールファイルマージをトリガーしたかどうかを確認できます。サブクエリの多段階実行と同様に、マージジョブも別のタブとして表示されます。自動スモールファイルマージのための追加の MergeTask は、現在のジョブの全体的な実行時間を増加させますが、マージ後の結果テーブルで生成されるファイルの数とサイズを最適化します。これにより、ファイルシステムへの過剰な圧力を防ぎ、後続のジョブでテーブルが使用される際の読み取りパフォーマンスを向上させます。マージジョブのタブ名は SQL_0_0_0_merge のようになり、MergeTask の実行進捗が表示されます。

  過剰なスモールファイルが存在する場合、ジョブが終了段階にあるときに SELECT ステートメントが長時間実行されることがあります。システムが SELECT ステートメントの実行結果を生成して表示するとき、システムは多数のスモールファイルを開いてファイルからデータを読み取る必要があります。このプロセスは時間がかかります。システムが大量の実行結果を生成するのを防ぐために、SELECT ステートメントを使用しないことを推奨します。Tunnel コマンドを使用してデータをダウンロードできます。実行結果の数は多くないが、ファイルの数が過剰に多い場合は、odps.merge.smallfile.filesize.threshold パラメーターが適切に設定されているかどうかを確認することを推奨します。スモールファイルのマージ方法の詳細については、「スモールファイルのマージ」をご参照ください。

• 動的パーティションでのメタデータ更新

  問題の説明：Fuxi ジョブが完了した後、メタデータに関連する特定の操作を実行する必要がある場合があります。たとえば、結果データを特定のディレクトリに移動し、テーブルのメタデータを更新する場合、動的パーティショニング中にテーブルに多数のパーティションが生成される可能性があります。その結果、プロセスに時間がかかります。たとえば、パーティションテーブル sales に対して insert into ... values ステートメントが実行され、2,000 個のパーティションが追加されます。サンプルステートメント：

  INSERT INTO TABLE sales partition (ds)(ds, product, price)
  VALUES ('20170101','a',1),('20170102','b',2),('20170103','c',3), ...;

  Fuxi ジョブが終了した後も、テーブルのメタデータを更新するには時間がかかります。LogView の [SubStatusHistory] は、ジョブが SQLTask is updating meta information でスタックしていることを示します。対応する状態コードは 1260 であり、その [Latency] の値はメタデータ更新にかかった時間を示します。

• 出力ファイルのサイズの増加

  現象：入出力レコードの数は似ていますが、出力サイズが数倍大きくなっています。[Fuxi Jobs] で、[IO bytes] 列を確認して、各タスクの入力データサイズと出力データサイズを比較します。これにより、サイズ肥大化が発生した段階を特定できます。

  解決策：1 つの理由は、データ分布の変化です。テーブルにデータを書き込むとき、データは圧縮されます。圧縮アルゴリズムは、反復的なデータに対して最も効果的です。したがって、書き込みプロセス中に同一のデータが一緒に配置されると、高い圧縮率を達成できます。データ分布は、書き込み段階 (Fuxi ジョブの R12 段階) でデータがどのようにシャッフルおよびソートされるかによって異なります。この例では、最終的な SQL 操作は JOIN であり、JOIN キーは次のとおりです：

  on  t1.query = t2.query and t1.item_id=t2.item_id

  データの特性は、ほとんどの列がアイテムの属性であることを示しています。この場合、同じ item_id を持つアイテムのすべての列は同じです。したがって、クエリ内のすべてのアイテムはランダムな順序でソートされます。これにより、圧縮率が低くなります。次のサンプルコードは、結合シーケンスを変更する方法の例を示しています。変更後、データサイズは元のデータサイズの 1/3 に減少します。

  on t1.item_id=t2.item_id and t1.query = t2.query

  変更後、データサイズは元のデータサイズの 1/3 に減少します。

  別のケースでは、JOIN または GROUP BY によって生成されるシャッフル操作に、圧縮に最も適したソート列が含まれていません。この場合、ZORDER BY 句を使用してオンプレミスマシンでアイテムをソートできます。これにより、より低いコストで高い圧縮率を得ることができます。DISTRIBUTED BY SORT BY ステートメントを実行して、データ分布を手動で再配置することもできます。この方法では、データ計算に長時間を要し、CPU 使用率が高くなります。