誰もが知っているように、世界のデータ量は増え続けており、情報の流れを収集して処理することがますます困難になっています。 これを行うには、MapReduceパラダイムを使用して、マルチスレッドアプリケーションの開発とデバッグの方法を簡素化するという考えで、人気のあるHadoopソリューションを使用します。 このパラダイムは常にそのタスクで成功するとは限りません。しばらくすると、Hadoopを超える「上部構造」があります： DAGパラダイムを備えたApache Tez 。 Tezの外観は、Hive HDFS-SQLハンドラーにも適応します。 しかし、常に新しいものが古いものより優れているわけではありません。 ほとんどの場合、HiveOnTezはHiveOnMapReduceよりも大幅に高速ですが、いくつかの落とし穴はソリューションのパフォーマンスに大きく影響します。 ここで、私が遭遇したニュアンスをお伝えしたいと思います。 これがETLまたは別のHadoop UseCaseの高速化に役立つことを願っています。

MapReduce、Tez、Hive

先ほど言ったように、世界にはますます多くのデータがあります。 また、ストレージと処理のために、Hadoopなどのますますトリッキーなソリューションが登場します。 HDFSに保存されたデータを処理するプロセスを平均的なアナリストでも簡単に行えるようにするために、HadoopにはいくつかのSQLアドオンがあります。 それらの中で最も古く「単純な」ものはHiveです。 Hiveの本質は次のとおりです。わかりやすい列ストア形式のデータがあり、それらに関する情報をメタデータに入力し、いくつかの制限付きの標準SQLを記述し、問題を解決するMapReduceジョブのチェーンを生成します。 素晴らしく、快適ですが、遅いです。 たとえば、簡単なクエリを次に示します。

select t1.column1, t2.column2 from table1 t1 inner join table2 t2 on t1.column1 = t2.column1 union select t3.column1, t4.column2 from table3 t3 inner join table4 t4 on t3.column1 = t4.column1 order by column1;
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

このクエリは4つのジョブを生成します。

• table1内部結合table2;
• table3内部結合table4;
• 組合;
• 並べ替えます。

ステップは順次実行され、各ステップはHDFSにデータを書き込むことで終了します。 非常に最適に見えません。 たとえば、ステップ1と2は並行して実行できます。 また、複数のステップで同じマッパーを適用し、これらのマッパーの結果に複数のタイプのレデューサーを適用することが賢明な場合があります。 しかし、1つのジョブのフレームワーク内でのMapReduceの概念では、これは許可されていません。 この問題を解決するために、DAGコンセプトのApache Tezがすぐに登場します。 DAGの本質は、Mapper-Reducerのペア（+イプシロン）の代わりに、非巡回有向グラフを作成することです。各頂点はMapper.ClassまたはReduser.Classであり、エッジはデータフロー/実行順序を示します。 TAGはDAGに加えて、さらにいくつかのボーナスを提供しました。ジョブの起動の高速化（既に実行中のTez-Engineを介してDAGジョブを送信できます）、ステップ間でノードメモリのリソースを保持する機能、独立して並列化を起動する機能などです。 Tezが登場し、対応するアドオンがHiveに追加されました。 このアドインを使用すると、クエリはほぼ次の構造のDAGジョブになります。

1. マッパーはtable1を読み取ります。
2. マッパーはtable2を読み取り、ステップ1の結果と結合します。
3. マッパーはtable3を読み取り、column1 IS NOT NULLをフィルターします。
4. マッパーはtable4を読み取り、column1 IS NOT NULLをフィルターします。
5. レデューサーは、ステップ3と4の結果を結合します。
6. レデューサーは結合を行います。
7. レデューサーのグループ化と並べ替え。
8. 結果を収集します。

実際、ステップ1と2は最初の結合であり、2、3と4は2番目の結合です（結合が異なる方法で処理されるように、異なるサイズのテーブルを特別に選択しました）。 この場合、2つのブロックは互いに独立しており、並行して実行できます。 これはすでに非常にクールです。 Tezは、複雑なクエリの処理速度を大幅に向上させます。 ただし、 set hive.execution.engine=tez



はMapReduceよりも悪い場合があるため、本番set hive.execution.engine=tez



に送信する前に、 set hive.execution.engine=mr



とset hive.execution.engine=mr



両方でクエリを実行する必要があります。

それで、テスとは何ですか？

Tezについて知っておくべきこと：MapReduceロジックをDAG（有向非巡回グラフ）に変更し、マッパーまたはリデューサーにかかわらず、同じDataFlow内でいくつかの異なるプロセスを同時に実行する機能を提供します。 主なことは、その入力が準備できていることです。 データは、ステップ間でノードにローカルに保存でき、ディスク操作に頼らずにノードのRAMに保存することもできます。 マッパーとリデューサーの数と場所を最適化して、マルチステップ計算を考慮してもネットワーク上のデータ転送を最小限に抑え、1つのTez-Jobのフレームワーク内で隣接プロセスで既に動作しているコンテナーを再利用し、統計に並列実行を調整できます。前のステップで収集されました。 さらに、このエンジンにより、エンドユーザーはMapReduceと同じシンプルさでDAGタスクを作成できますが、彼自身はリソース、再起動、クラスターのDAG管理に従事します。 Tezは非常にモバイルであり、Tezサポートを追加しても既に実行中のプロセスが中断されることはありません。また、古いバージョンのTezがすべてのクラスタータスクで動作する場合、ローカルで「クライアント側」で新しいバージョンをテストできます。 最後になりましたが、Tezはクラスター上でサービスとして実行し、バックグラウンドで実行できるため、MapReduceが正常に起動されたときよりもはるかに高速にタスクを送信できます。 Tezを試したことがなく、まだ疑問がある場合は、 HortonWorks プレゼンテーションで公開されている速度比較をご覧ください。







そして、Hiveとペアになりました：







しかし、グラフと説明のすべての美しさには、HiveOnTezに問題があります。

Tezは、MapReduceよりも不均一なデータ分布に対する耐性が低い

最初の最大の問題は、DAG-jobとMapReduce-jobの作成の違いにあります。 それらには1つの原則があります。マッパーとリデューサーの数は、ジョブの開始時に計算されます。 MapReduce-jobsのチェーンによってクエリが実行される場合にのみ、Hadoopは前のステップの結果とソースによって収集された分析に基づいて必要なタスク数を計算します。DAG-jobの場合、これはすべてのステップが分析に基づいてのみ計算される前に行われます。



例で説明します。 クエリの途中のどこかで、ネストされたクエリを実行すると、2つのテーブルがあります。 統計によると、それぞれにはn行とk個の一意の結合キー値があります。 出力では、約n * k行が予想されます。 そして、この数量が1つのコンテナにうまく収まり、Tezが次のステップ（ソートなど）で1つのReducerを強調表示するとします。 そして、このReducerの数は、何があっても実行プロセス中に変化しません。 ここで、実際にはこれらのテーブルのスキューが非常に悪いと仮定します。1つの値に対してn-k + 1行があり、残りはすべて1行に対してです。 したがって、出力ではn ^ 2 + k ^ 2-2kn-k + 2n行になります。 つまり、（n + 2-2k）/ k +（k-1）/ nはn / kの2倍の大きさです。 そして、そのような量の1つのReducerは永遠を実行します。 MapReduceの場合、このステップの出力でn ^ 2 + k ^ 2-2kn-k + 2nを受け取ったHadoopは、その強度を客観的に評価し、必要な数のMapperとReducerを提供します。 その結果、MapReduceを使用すると、すべてがはるかに高速に動作します。



ドライな計算は非常に手間がかかるように見えるかもしれませんが、実際にはこの状況は現実です。 そして、それが起こらなかった場合、あなたは幸運だと考えてください。 複雑なクエリまたはカスタムマッパーでラテラルビューを使用すると、同様のTez-DAG効果に遭遇しました。

Tezのチューニング機能

皮肉なことに、私が知っている最後の重要なTez機能は、そのDAGパワーです。 ほとんどの場合、クラスターは単なる情報のリポジトリではありません。 また、データが処理されるシステムでもあり、アクティビティの残りがクラスタのこの部分に影響を与えないことが重要です。 ノードはリソースであるため、通常、コンテナの数は無制限ではありません。 したがって、ジョブを実行するときは、通常のプロセスを大幅に遅くしないために、すべてのコンテナを詰まらせない方が良いです。 そして、ここでDAGはあなたにブタを置くことができます。 DAGでは、再利用やスムーズな装填などにより、必要なコンテナの数が（チャンバーあたり平均）少なくなります。しかし、多くのクイックステップがあると、コンテナは指数関数的に増加し始めます。 最初のマッパーはまだ完成していませんが、データはすでに他のマッパーに配布されており、コンテナはすべてこれに割り当てられています-ブーム！ クラスターが天井に詰まっているため、他の誰も単一のジョブを開始できません。 十分なリソースがなく、進行状況バーの数値がどれだけゆっくり変化するかを確認します。 一貫性があるため、MapReduceはこの効果を免れますが、いつものように、高速で料金を支払います。



標準のMapReduceがあまりにも多くのコンテナを占有するという事実に対処する方法を長い間知っています。 パラメーターを調整します。

• mapreduce.input.fileinputformat.split.maxsize
  
  
  
  ：減少-マッパーの数を増やします。
• mapreduce.input.fileinputformat.split.minsize
  
  
  
  ：増加-マッパーの数を減らします。
• mapreduce.input.fileinputformat.split.minsize.per.node
  
  
  
  、 mapreduce.input.fileinputformat.split.minsize.per.rack
  
  
  
  ：ローカル（ノードまたはラックの意味で）パーティションを制御するためのより細かい設定。
• hive.exec.reducers.bytes.per.reducer
  
  
  
  ：増加- hive.exec.reducers.bytes.per.reducer
  
  
  
  の数を減らします。
• mapred.tasktracker.reduce.tasks.maximum
  
  
  
  ： mapred.tasktracker.reduce.tasks.maximum
  
  
  
  の最大数を設定します。
• mapred.reduce.tasks
  
  
  
  ：特定の数のmapred.reduce.tasks
  
  
  
  設定します。

ご注意 DAGでは、すべてのリデュースステップに、ここで指定した数のプロセスが含まれます！ ただし、Tezパラメーターはより複雑であり、MapReduceに設定したパラメーターが常に影響するわけではありません。 まず、 hive.tez.container.size



に非常に敏感であり、インターネットはyarn.scheduler.minimum-allocation-mb



とyarn.scheduler.maximum-allocation-mb



間の値を取ることをyarn.scheduler.minimum-allocation-mb



ています。 次に、未使用のコンテナーの保持パラメーターを確認します。

• tez.am.container.ide.release-timeout-max.millis
  
  
  
  ;
• tez.am.container.ide.release-timeout-min.millis
  
  
  
  。

tez.am.container.reuse.enabled



オプションは、コンテナの再利用を有効または無効にします。 無効にすると、前の2つのパラメーターは機能しません。 そして第三に、グループ化オプションを見てください。

• tez.grouping.split-waves
  
  
  
  ;
• tez.grouping.max-size
  
  
  
  ;
• tez.grouping.min-size
  
  
  
  。

実際には、外部データの読み取りを並列化するために、Tezはタスクを形成するプロセスを変更しました。最初に、 tez.grouping.split-waves



はクラスターで実行できる波数（w）を推定し、次にこの数にtez.grouping.split-waves



パラメーターを乗算し、積（N）を分割しますタスクごとの標準分割数。 アクションの結果がtez.grouping.min-size



とtez.grouping.max-size



間にある場合、すべてが正常であり、タスクはN個のタスクで開始されます。 そうでない場合、番号はフレームに適合します。 Tezのドキュメントでは、「実験としてのみ」 tez.grouping.split-count



パラメーターを設定することをtez.grouping.split-count



ます。これにより、上記のすべてのロジックがキャンセルされ、パラメーターで指定されたグループ数に分割がグループ化されます。 しかし、私はこのプロパティを使用しないようにします。特定の入力データを最適化するためにTezとHadoop全体に柔軟性を与えません。

テュスのニュアンス

大きな問題に加えて、テズは小さな欠陥を免れません。 たとえば、http Hadoop ResourceManagerを使用する場合、Tez-jobがコンテナーを占有する量は表示されません。さらに、マッパーとリデューサーの状態は表示されません。 クラスターの状態を監視するには、次の小さなPythonスクリプトを使用します。

 import os import threading result = [] e = threading.Lock() def getContainers(appel): attemptfile = os.popen("yarn applicationattempt -list " + appel[0]) attemptlines = attemptfile.readlines() attemptfile.close() del attemptlines[0] del attemptlines[0] for attempt in attemptlines: splt = attempt.split('\t'); if ( splt[1].strip() == "RUNNING" ): containerfile = os.popen("yarn container -list " + splt[0] ) containerlines = containerfile.readlines() containerfile.close() appel[2] += int( containerlines[0].split("Total number of containers :")[1].strip() ) e.acquire() result.append(appel) e.release() appfile = os.popen("yarn application -list -appStates RUNNING") applines = appfile.read() appfile.close() apps = applines.split('application_') del apps[0] appsparams = [] for app in apps: splt = app.split('\t') appsparams.append(['application_' + splt[0],splt[3], 0]) cnt = 0 threads = [] for app in appsparams: threads.append(threading.Thread(target=getContainers, args=(app,))) for thread in threads: thread.start() for thread in threads: thread.join() result.sort( key=lambda x:x[2] ) total = 0 for app in result: print(app[0].strip() + '\t' + app[1].strip() + '\t' + str(app[2]) ) total += app[2] print("Total:",total)
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

HortonWorksの保証にもかかわらず、私たちのプラクティスは、Hiveで単純なSELECT smth FROMテーブルWHERE smthを実行すると、ほとんどの場合MapReduceがより速く動作することを示しています。 さらに、記事の冒頭で、私はあなたを少しだましました。HiveOnMapReduceでの並列化は可能ですが、それほどスマートではありません。 必要なことは、 set hive.exec.parallel=true



にset hive.exec.parallel.thread.number=



...にset hive.exec.parallel.thread.number=



です-そして、独立したステップ（マッパー+レデューサーペア）が並列に実行されます。 はい、1つのMapperの出力で複数のReducerまたは次のMapperが起動される可能性はありません。 はい、並列化ははるかに原始的ですが、作業を高速化します。



Tezのもう1つの興味深い機能は、クラスターでエンジンを実行し、しばらくエンジンを保持することです。 一方では、タスクがノード上ではるかに高速に実行されるため、これにより作業が本当に高速化されます。 ただし、一方で、予期しないマイナスがあります。重要なプロセスは、このモードでは開始できません。TEZエンジンは、時間の経過とともに多くのクラスを生成し、GCオーバーフローでクラッシュするためです。 そして、それは次のようになります： nohup hive -f ....hql > hive.log &



し、夜になって午前中に来ました。 不快です。



古き良きMapReduceが既に安定版リリースに含まれているという小さな問題の貯金箱に追加され、TEZは人気と進歩性にもかかわらず、バージョン0.8.4のままであり、どの段階でもバグに遭遇する可能性があります。 私にとって最悪のバグは情報の削除ですが、そのようなことは見たことがありません。 しかし、Tezで誤った計算が発生し、MapReduceはそれを正しいと見なします。 たとえば、私の同僚は、一意のEntityIdフィールドを持つ2つのテーブルtable1とtable2を使用しました。 Tezを通じてリクエストを行いました。

 select table1.EntityId, count(1) from table1 left join table2 on table1.EntityId = table2.EntityId group by EntityId having count(1) > 1
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

そして、出力にいくつかの行がありました！ ただし、MapReduceは空の結果を返します。 同様の問題に関するバグレポートがあります。

おわりに

Tezは無条件の利点であり、ほとんどの場合、生活が楽になり、Hiveでより複雑なクエリを記述し、それらに対する迅速な回答を期待できます。 しかし、他の善と同様に、慎重なアプローチ、慎重さ、およびいくつかのニュアンスの知識が必要です。 その結果、古い、実績のある、信頼できるMapReduceを使用する方がTezを使用するよりも優れている場合があります。 HiveOnTezのマイナスに関する記事（RuNetでも英語でもない）が1つも見つからなかったことに非常に驚き、このギャップを埋めることに決めました。 この情報が誰かに役立つことを願っています。 よろしくお願いします！ みなさん、さようなら！