MaxCompute:DISTRIBUTED MAPJOIN

When to use DISTRIBUTED MAPJOIN

DISTRIBUTED MAPJOIN is designed for joins where the table sizes differ significantly:

• The large table exceeds 10 TB.

• The small table is between 1 GB and 100 GB.

• Data in the small table is evenly distributed (no significant skew).

If an SQL task takes more than 20 minutes to complete, consider using DISTRIBUTED MAPJOIN for optimization.

If the small table has long-tail data (skewed distribution), a single shard may receive disproportionate data. This can cause an out-of-memory (OOM) error or a remote procedure call (RPC) timeout.

How it works

DISTRIBUTED MAPJOIN splits the small table into a configurable number of shards and distributes them across compute nodes. Each node processes its assigned shards against the corresponding portion of the large table. This avoids loading the entire small table onto every node.

Task parallelism is determined by the following formula:

Parallelism = shard_count x replica_count

Syntax

Add a hint to your SELECT statement:

/*+distmapjoin(<table_name>(shard_count=<n>,replica_count=<m>))*/

-- Specify shard_count only (replica_count defaults to 1)
/*+distmapjoin(a(shard_count=5))*/

-- Specify both shard_count and replica_count
/*+distmapjoin(a(shard_count=5,replica_count=2))*/

-- Join multiple small tables with DISTRIBUTED MAPJOIN
/*+distmapjoin(a(shard_count=5,replica_count=2),b(shard_count=5,replica_count=2))*/

-- Combine DISTRIBUTED MAPJOIN with MAPJOIN
/*+distmapjoin(a(shard_count=5,replica_count=2)),mapjoin(b)*/

Use DISTRIBUTED MAPJOIN in a query

Before you begin, make sure that you have:

• A MaxCompute project with sufficient quota resources

• Tables that meet the size requirements (large table > 10 TB, small table between 1 GB and 100 GB)

DISTRIBUTED MAPJOIN is resource-intensive. Avoid running tasks in a small quota group. Change the quota group on the Quotas page. For more information, see Manage quotas in the new MaxCompute console.

The following examples insert data into a partitioned table named tmall_dump_lasttable. The large table is search_ods.dump_lasttable and the small table is a filtered subset of tbcdm.dim_tb_itm.

Standard 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;

With DISTRIBUTED MAPJOIN:

Add the distmapjoin hint to the SELECT statement. In this example, the small table t2 is split into 35 shards:

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;

Best practices

Troubleshooting