MaxCompute:DISTRIBUTED MAPJOIN

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

• パラメーター

  • table_name: 結合する小さなテーブルの名前。

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

    説明

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

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

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

    説明

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

• 構文

  -- Recommended. Specify the shard_count parameter and retain the default value 1 for the replica_count parameter.
  /*+distmapjoin(a(shard_count=5))*/
  
  -- Recommended. Specify the shard_count and replica_count parameters.
  /*+distmapjoin(a(shard_count=5,replica_count=2))*/
  
  -- Use DISTRIBUTED MAPJOIN to join multiple small tables.
  /*+distmapjoin(a(shard_count=5,replica_count=2),b(shard_count=5,replica_count=2)) */
  
  -- Use DISTRIBUTED MAPJOIN and MAPJOIN together.
  /*+distmapjoin(a(shard_count=5,replica_count=2)),mapjoin(b)*/

この例では、tmall_dump_lasttableという名前のパーティションテーブルにデータを挿入するときにDISTRIBUTED 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='20211203'
  )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;

• DISTRIBUTED 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='20211203'
  )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;