MaxCompute:分散 MAPJOIN

SELECT 文で DISTRIBUTED MAPJOIN を使用するには、文にヒント /*+distmapjoin(<table_name>(shard_count=<n>,replica_count=<m>))*/ を追加する必要があります。shard_count と replica_count パラメーターは両方とも、タスクの並列度を決定するために使用します。並列度は、次の式で計算できます：並列度 = shard_count × replica_count。

• パラメーター

  • table_name：結合する小さいテーブルの名前です。

  • shard_count=<n>：結合する小さいテーブルのデータシャードの数です。小さいテーブルのデータシャードは、データ処理のために各コンピューティングノードに分散されます。n はシャードの数を指定します。ほとんどの場合、このパラメーターには奇数を設定します。

    説明

    • shard_count パラメーターは手動で指定することを推奨します。shard_count パラメーターの値は、小さいテーブルのサイズに基づいて見積もることができます。単一のシャードノードで処理されるデータ量の推定値は、[200 MB, 500 MB] の範囲内です。

    • shard_count パラメーターを大きすぎる値に設定すると、データ処理のパフォーマンスと安定性に悪影響を及ぼします。shard_count パラメーターを小さすぎる値に設定すると、メモリリソースの過剰な使用が原因でエラーが発生する可能性があります。

  • replica_count=<m>：小さいテーブルのレプリカの数です。m はレプリカの数を指定します。デフォルト値：1。

    説明

    過剰なアクセスリクエストを削減し、単一ノードの障害によるタスク全体の失敗を防ぐために、同じシャード内のデータのレプリカを複数作成できます。タスクの並列度が高い、または環境のパフォーマンスが不安定なためにノードが頻繁に再起動する場合は、replica_count パラメーターの値を増やすことができます。このパラメーターは 2 または 3 に設定することを推奨します。

• 構文

  -- 推奨：shard_count を指定します。replica_count のデフォルトは 1 です。
  /*+distmapjoin(a(shard_count=5))*/
  
  -- 推奨：shard_count と replica_count の両方を指定します。
  /*+distmapjoin(a(shard_count=5,replica_count=2))*/
  
  -- 複数の小さいテーブルに分散 MAPJOIN を使用します。
  /*+distmapjoin(a(shard_count=5,replica_count=2),b(shard_count=5,replica_count=2)) */
  
  -- 分散 MAPJOIN と MAPJOIN を一緒に使用します。
  /*+distmapjoin(a(shard_count=5,replica_count=2)),mapjoin(b)*/

この例では、パーティションテーブル tmall_dump_lasttable にデータを挿入する際に、分散 MAPJOIN を使用する方法を説明します。

• JOIN 構文

  insert OVERWRITE table tmall_dump_lasttable partition(ds='20211130')
  select t1.*
  from
  (
      select nid, doc,type
      from search_ods.dump_lasttable where ds='20211130'
  )t1
  join
  (
      select distinct item_id
      from tbcdm.dim_tb_itm
      where ds='20211130'
      and bc_type='B'
      and is_online='Y'
  )t2
  on t1.nid=t2.item_id;

• 分散 MAPJOIN 構文

  insert OVERWRITE table tmall_dump_lasttable partition (ds='20211130')
  select /*+ distmapjoin(t2(shard_count=35)) */ t1.*
  from
  (
      select nid, doc, type
      from search_ods.dump_lasttable where ds='20211130'
  )t1
  join
  (
      select distinct item_id
      from tbcdm.dim_tb_itm
      where ds='20211130'
      and bc_type='B'
      and is_online='Y'
  )t2
  on t1.nid=t2.item_id;