MaxCompute は、SQL ランタイム実行モードでの MapReduce ジョブの実行をサポートしています。SQL ランタイムを使用することで、MapReduce ジョブは SQL エンジンの新機能を活用し、以前はサポートされていなかった機能を実現できます。このトピックでは、MapReduce ジョブに SQL ランタイム実行モードを指定する方法について説明します。
背景情報
MaxCompute は、MaxCompute 内のデータを処理する Java プログラムを作成するための MapReduce API を提供します。
最新バージョンの MaxCompute では、MapReduce ジョブを SQL ランタイム実行モードで実行できます。このモードでは、MapReduce ジョブは MaxCompute SQL コンパイラ、コストベースオプティマイザ (CBO)、およびベクトル化実行エンジンを使用でき、機能、パフォーマンス、安定性における SQL エンジンの継続的な改善の恩恵を受けることができます。
SQL ランタイム実行モードで実行される MapReduce ジョブは、MaxCompute SQL の新機能を使用して、以前はサポートされていなかった操作を実行できます。従来の MapReduce エンジンと比較して、SQL ランタイム実行モードには次の機能が追加されています。
-
入力ソースとしてのビューのサポート。
-
入力ソースとしての外部テーブルのサポート。
-
分散ファイルシステムでの読み取りおよび書き込み操作のサポート。
-
ハッシュクラスター化テーブルまたはレンジクラスター化テーブルでの読み取りおよび書き込み操作のサポート。
このモードは、次の利点も提供します。
-
SQL コストベースオプティマイザ (CBO) とベクトル化実行エンジンの継続的なパフォーマンス最適化を活用できます。
-
新しいストレージ形式の圧縮メカニズムを使用できます。
-
ハッシュクラスター化テーブルなどの非常に大きなテーブルを結合するシナリオで、並列度を動的に調整してパフォーマンスを向上させることができます。
-
広範なジョブ実行とストレステストによって実証された、SQL エンジンの安定性を活用できます。これにより、フェールオーバーや永続ボリューム要求 (PVC) などのメカニズムの信頼性が向上します。
-
MaxCompute Studio と LogView を使用して、より深いインサイトを得ることができます。これらのツールは、実行計画、コンパイル情報、ジョブ設定など、SQL ジョブの詳細な実行情報を提供します。各フェーズでの I/O と全体的なワークフローを詳細に表示することで、問題の特定、パフォーマンスの最適化、開発と運用保守効率の向上に役立ちます。
注意
-
この機能では、既存の API やジョブログロジックを変更する必要はありません。実行モードを指定するだけで済みます。
-
この機能は、MapReduce API で作成された MapReduce ジョブのみをサポートします。詳細については、「SDK の概要」をご参照ください。
-
SQL ランタイム実行モードで実行される MapReduce ジョブは、引き続き標準の MapReduce 課金ルールに従って課金されます。詳細については、「」「MapReduce 従量課金」をご参照ください。
手順
-
実行モードを設定します。
odps.mr.run.modeプロパティで実行モードを制御できます。有効な値は次のとおりです：-
lot(デフォルト)：ジョブは MapReduce 実行エンジンで実行されます。 -
sql：ジョブは SQL 実行エンジンで実行されます。ジョブが失敗した場合、エラーが返されます。 -
hybrid：システムはまず SQL 実行エンジンでジョブを実行しようとします。試行が失敗した場合、ジョブは MapReduce 実行エンジンにフォールバックします。
実行モードは、次のいずれかの方法で設定できます：
プロジェクトレベルでの制御
プロジェクト内のすべてのジョブでこのモードを有効にするには、プロジェクト管理者が次のコマンドを実行する必要があります：
setproject odps.mr.run.mode=<lot/sql/hybrid>;セッションレベルでの制御
現在のジョブでのみこのモードを有効にするには、次のいずれかの方法を使用します：
-
JAR コマンドの前に
set odps.mr.run.mode=<lot/sql/hybrid>ステートメントを追加します。 -
次の例に示すように、JobConf オブジェクトを使用してジョブコードでプロパティを設定します：
JobConf job = new JobConf(); job.set("odps.mr.run.mode","hybrid")
説明StreamJob や SecondarySort などの特別なシナリオでは、次のフラグを設定する必要があります：
-
StreamJob：
set odps.mr.sql.stream.enable=true; -
SecondarySort：
set odps.mr.sql.group.enable=true;
-
-
ジョブの詳細を表示します。
LogView や MaxCompute Studio などのツールを使用して、クライアントで生成された SQL 式を検査し、ジョブの実行詳細を表示できます。LogView の使用方法の詳細については、「LogView V2.0 を使用したジョブ情報の表示」をご参照ください。
-
LogView XML
LogView を開いて [ソース XML] タブに移動すると、クライアントから送信された XML 情報を表示できます。この情報には、MapReduce ジョブと等価な SQL が示されています。以下にコード例を示します。
create temporary function mr2sql_mapper_152955927079392291755 as 'com.aliyun.odps.mapred.bridge.LotMapperUDTF' using ; create temporary function mr2sql_reducer_152955927079392291755 as 'com.aliyun.odps.mapred.bridge.LotReducerUDTF' using ; @sub_query_mapper := SELECT k_id,v_gmt_create,v_gmt_modified,v_product_id,v_admin_seq,v_sku_attr,v_sku_price,v_sku_stock,v_sku_code,v_sku_image,v_delivery_time,v_sku_bulk_order,v_sku_bulk_discount,v_sku_image_version,v_currency_code FROM( SELECT mr2sql_mapper_152955927079392291755(id,gmt_create,gmt_modified,product_id,admin_seq,sku_attr,sku_price,sku_stock,sku_code,sku_image,delivery_time,sku_bulk_order,sku_bulk_discount,sku_image_version,currency_code ) as (k_id,v_gmt_create,v_gmt_modified,v_product_id,v_admin_seq,v_sku_attr,v_sku_price,v_sku_stock,v_sku_code,v_sku_image,v_delivery_time,v_sku_bulk_order,v_sku_bulk_discount,v_sku_image_version,v_currency_code) FROM ae_antispam.product_sku_tt_inc WHERE ds = "20180615" AND hh = "21" UNION ALL SELECT mr2sql_mapper_152955927079392291755(id,gmt_create,gmt_modified,product_id,admin_seq,sku_attr,sku_price,sku_stock,sku_code,sku_image,delivery_time,sku_bulk_order,sku_bulk_discount,sku_image_version,currency_code ) as (k_id,v_gmt_create,v_gmt_modified,v_product_id,v_admin_seq,v_sku_attr,v_sku_price,v_sku_stock,v_sku_code,v_sku_image,v_delivery_time,v_sku_bulk_order,v_sku_bulk_discount,v_sku_image_version,v_currency_code) FROM ae_antispam.product_sku ) open_mr_alias1 DISTRIBUTE BY k_id SORT BY k_id ASC; @sub_query_reducer := SELECT mr2sql_reducer_152955927079392291755(k_id,v_gmt_create,v_gmt_modified,v_product_id,v_admin_seq,v_sku_attr,v_sku_price,v_sku_stock,v_sku_code,v_sku_image,v_delivery_time,v_sku_bulk_order,v_sku_bulk_discount,v_sku_image_version,v_currency_code) as (id,gmt_create,gmt_modified,product_id,admin_seq,sku_attr,sku_price,sku_stock,sku_code,sku_image,delivery_time,sku_bulk_order,sku_bulk_discount,sku_image_version,currency_code) FROM @sub_query_mapper; FROM @sub_query_reducer INSERT OVERWRITE TABLE ae_antispam.product_sku SELECT id,gmt_create,gmt_modified,product_id,admin_seq,sku_attr,sku_price,sku_stock,sku_code,sku_image,delivery_time,sku_bulk_order,sku_bulk_discount,sku_image_version,currency_code ; -
LogView の概要
LogView の [概要] タブで、ジョブが SQL
execution engineを使用して実行されたことを確認できます。次に例を示します：説明SQL ランタイム実行モードで実行されない MapReduce ジョブの場合、実行エンジン情報は表示されません。SQL ランタイム実行モードで実行されない MaxCompute 拡張 MapReduce (MR2) ジョブの場合、実行エンジンは
cganjiangです。Job run mode: fuxi job Job run engine: execution engine -
LogView の JSON 概要
標準の MapReduce ジョブの JSON サマリーには、Map タスクと Reduce タスクに関する基本的な入出力情報のみが含まれます。一方、SQL ベースのジョブの JSON サマリーでは、すべての実行パラメーター、論理プラン、物理プラン、実行詳細など、各実行フェーズの詳細な情報が提供されます。以下のコードに例を示します。
"midlots" : [ "LogicalTableSink(table=[[odps_flighting.flt_20180621104445_step1_ad_quality_tech_qp_algo_antifake_wordbag_filter_bag_change_result_lv2_20, auctionid,word,match_word(3) {0, 1, 2}]]) OdpsLogicalProject(auctionid=[$0], word=[$1], match_word=[$2]) OdpsLogicalProject(auctionid=[$0], word=[$1], match_word=[$2]) OdpsLogicalProject(auctionid=[$0], word=[$1], match_word=[$2]) OdpsLogicalProject(auctionid=[$2], word=[$3], match_word=[$4]) OdpsLogicalTableFunctionScan(invocation=[[MR2SQL_MAPPER_152955294118813063732($0, $1)]()], rowType=[RecordType(VARCHAR(2147483647) item_id, VARCHAR(2147483647) text, VARCHAR(2147483647) __tf_0_0, VARCHAR(2147483647) __tf_0_1, VARCHAR(2147483647) __tf_0_2)]) OdpsLogicalTableScan(table=[[ad_quality_tech.qp_algo_antifake_wordbag_filter_bag_change_lv2_20, item_id,text(2) {0, 1}]]) ]
-