Map-side join and the distributed cache

Give every worker a copy of the menu card

Map-side join and the distributed cache

Words you'll meet in this lesson

From our lab (Hadoop 3.5.0, 5 lakh orders, menu without “coffee”)

Map-side join at three levels

Reduce-side vs map-side joins

Three ways to put a file in the distributed cache

A replicated join on the lab cluster

Same data as Lesson 9.5: student_orders.csv (5 lakh orders, 3 splits) and a menu with “coffee” removed on purpose. Full scripts are in “Run it on your PC”.

1. Register the small file and switch off reducers

   job.addCacheFile(new URI(rest[1] + "#menu"));   // ship the menu, linked as ./menu
   job.setMapperClass(JoinMapper.class);
   job.setNumReduceTasks(0);                       // no reducer, so no shuffle at all

   The client records the HDFS path in the job configuration. Before any map container starts on a node, that node's NodeManager downloads the file once into its local file cache.

2. Load it once per task, in setup()

   try (BufferedReader r = new BufferedReader(new FileReader("menu"))) {
     String line;
     while ((line = r.readLine()) != null) {
       String[] f = line.split(",");
       category.put(f[0], f[1]);
     }
   }

   cache files: [/user/hadoop/data/menu_nocoffee.csv#menu] menu loaded in setup(): 7 items, working dir = /tmp/hadoop-hadoop/nm-local-dir/usercache/hadoop/appcache/application_1792643020992_0040/container_1792643020992_0040_01_000003/.
   The file is opened by its plain name menu: YARN placed a symlink with that name in the container's working directory, pointing to the localized copy. No HDFS client code is needed.

3. Join while reading

   String cat = category.get(line.toString().split(",")[3]);
   if (cat == null) { ctx.getCounter(Join.ORDERS_WITHOUT_MENU).increment(1); return; }
   out.set(line + "," + cat);
   ctx.write(out, NullWritable.get());

   One hash lookup per order. Orders with no menu row (all coffee orders) are counted and dropped: an inner join. Writing them with an empty category instead would make it a left outer join.

4. Run it

   hadoop jar mr9.jar JoinMapSide /user/hadoop/data/student_orders.csv /user/hadoop/data/menu_nocoffee.csv /out/JoinMapSide

   number of splits:3 Launched map tasks=3 Map input records=500000 Map output records=459636 ORDERS_JOINED=459636 ORDERS_WITHOUT_MENU=40364 Bytes Written=20851062 took 18943 ms
   No “Launched reduce tasks”, no shuffle counters, no Spilled Records: map output goes straight to HDFS. The reduce-side join of the same data needed a reducer and moved 24,448,645 bytes.

5. Look at the output

   hdfs dfs -ls /out/JoinMapSide
   hdfs dfs -cat /out/JoinMapSide/part-m-00000 | head -2

   9020852 /out/JoinMapSide/part-m-00000 8993924 /out/JoinMapSide/part-m-00001 2836286 /out/JoinMapSide/part-m-00002 1,S1413,2026-09-14 15:30,dosa,45,meal 2,S0973,2026-09-26 20:11,samosa,18,snack
   One file per map task, rows in the same order as the input (order 1, then order 2). The reduce-side output came out grouped by item instead, because the reducer received it sorted by key.

6. Check it against the reduce-side join

   diff <(hdfs dfs -cat /out/JoinReduceSide/part-* | sort) <(hdfs dfs -cat /out/JoinMapSide/part-* | sort)

   identical (459,636 rows each, compared after sorting)
   Same answer, different route. Always verify a rewritten join this way before trusting its speed.

From HDFS to every container, then a lookup per order

Six steps of the JoinMapSide job: register, localize, symlink, load, join, write.

Map-side joins at work

Simplified; organisations differ in tools and policies.

Simulate a broadcast join, split by split

Every “map task” below gets its own copy of the small table, builds a dictionary once (its setup()), and joins its own split. There is no shuffle step at all, and each task's output keeps its input order.

menu_lines = ["chai,drink,15,A", "samosa,snack,18,B", "dosa,meal,45,D", "idli,breakfast,30,C"]
splits = [["1,S1413,2026-09-14 15:30,dosa,45", "2,S0973,2026-09-26 20:11,samosa,18"],
          ["3,S0007,2026-09-01 13:37,coffee,20", "4,S0002,2026-09-03 08:10,chai,15"],
          ["5,S0042,2026-09-09 14:05,idli,30"]]
joined = without_menu = 0
for n, split in enumerate(splits):
    category = {}                                   # setup(): once per map task
    for line in menu_lines:
        f = line.split(",")
        category[f[0]] = f[1]
    out = []
    for order in split:                             # map(): once per record
        cat = category.get(order.split(",")[3])
        if cat is None:
            without_menu += 1
            continue
        joined += 1
        out.append(order + "," + cat)
    print(f"part-m-0000{n}: {out}")
print("ORDERS_JOINED =", joined, " ORDERS_WITHOUT_MENU =", without_menu)

• category = {} # setup(): once per map taskIn Java this is the HashMap filled in setup() from the ./menu symlink: once per task, never per record.
• cat = category.get(order.split(",")[3])The join itself: one dictionary lookup. No shuffle, no sort, no reducer.
• without_menu += 1Coffee is missing from this menu, so its order is counted and dropped (an inner join). Our lab counted 40,364 of them.

Run it on your PC: map-side vs reduce-side join

Use the Module 8 lab cluster (lab8) with student_orders.csv and menu.csv from Lesson 9.1 (make_student_orders.py). Copy both Java files of Lessons 9.5 and 9.6 into /tmp/mr9 in the namenode container, compile them into mr9.jar, and run join_lab.sh. It removes “coffee” from the menu first so you can see unmatched orders being counted.

// JoinMapSide.java: map-side join, a map-only job. The small menu is shipped to every map task
// with the distributed cache and loaded into a HashMap in setup(); orders are joined as they are read.
import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.net.URI;
import java.util.HashMap;
import java.util.Map;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

public class JoinMapSide {
  enum Join { ORDERS_JOINED, ORDERS_WITHOUT_MENU }

  public static class JoinMapper extends Mapper<LongWritable, Text, Text, NullWritable> {
    private final Map<String, String> category = new HashMap<>();
    private final Text out = new Text();

    @Override
    protected void setup(Context ctx) throws IOException {
      URI[] files = ctx.getCacheFiles();
      System.err.println("cache files: " + java.util.Arrays.toString(files));
      // "#menu" in the URI makes YARN create a symlink called "menu" in the task's working directory
      try (BufferedReader r = new BufferedReader(new FileReader("menu"))) {
        String line;
        while ((line = r.readLine()) != null) {
          String[] f = line.split(",");
          category.put(f[0], f[1]);
        }
      }
      System.err.println("menu loaded in setup(): " + category.size() + " items, working dir = " + new java.io.File(".").getAbsolutePath());
    }

    @Override
    protected void map(LongWritable k, Text line, Context ctx) throws IOException, InterruptedException {
      String cat = category.get(line.toString().split(",")[3]);
      if (cat == null) { ctx.getCounter(Join.ORDERS_WITHOUT_MENU).increment(1); return; }
      ctx.getCounter(Join.ORDERS_JOINED).increment(1);
      out.set(line + "," + cat);
      ctx.write(out, NullWritable.get());
    }
  }

  public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    String[] rest = new GenericOptionsParser(conf, args).getRemainingArgs();   // orders menu out
    Job job = Job.getInstance(conf, "map-side join");
    job.setJarByClass(JoinMapSide.class);
    job.addCacheFile(new URI(rest[1] + "#menu"));     // ship the menu to every task, linked as ./menu
    job.setMapperClass(JoinMapper.class);
    job.setNumReduceTasks(0);                         // no reducer, so no shuffle at all
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(NullWritable.class);
    FileInputFormat.addInputPath(job, new Path(rest[0]));
    FileOutputFormat.setOutputPath(job, new Path(rest[2]));
    System.exit(job.waitForCompletion(true) ? 0 : 1);
  }
}

// Inside the namenode container, in /tmp/mr9:
//   mkdir -p classes && javac -cp $(hadoop classpath) -d classes *.java && jar cf mr9.jar -C classes .
//   hadoop jar mr9.jar JoinMapSide /user/hadoop/data/student_orders.csv /user/hadoop/data/menu.csv /out/JoinMapSide

# join_lab.sh: join 5 lakh orders with the 8-line menu, reduce-side and map-side, and compare
cd /tmp/mr9
O=/user/hadoop/data/student_orders.csv; MENU=/user/hadoop/data/menu.csv
hdfs dfs -put -f menu.csv $MENU
echo "== menu.csv (minus 'coffee', to show an order without a menu row)"
grep -v coffee menu.csv > menu_nocoffee.csv; hdfs dfs -put -f menu_nocoffee.csv /user/hadoop/data/
cat menu_nocoffee.csv
for j in JoinReduceSide JoinMapSide; do
  hdfs dfs -rm -r -f -skipTrash /out/$j > /dev/null 2>&1
  start=$(date +%s%N)
  hadoop jar mr9.jar $j $O /user/hadoop/data/menu_nocoffee.csv /out/$j > /tmp/$j.log 2>&1
  echo "== $j   $(( ($(date +%s%N) - start) / 1000000 )) ms"
  grep -E "number of splits|Launched (map|reduce) tasks|Map input records|Map output records|Reduce shuffle bytes|Reduce input groups|ORDERS_JOINED|ORDERS_WITHOUT_MENU|MENU_(NOT_)?FIRST|Bytes Written" /tmp/$j.log | sed -E 's/^[0-9-]+ [0-9:]+ INFO  //; s/^\s*/  /'
  hdfs dfs -ls /out/$j | grep part | awk '{print "  " $5, $8}'
done
echo "== a joined row"
hdfs dfs -cat /out/JoinMapSide/part-m-00000 2>/dev/null | head -2
echo "== same rows from both joins?"
diff <(hdfs dfs -cat /out/JoinReduceSide/part-* | sort) <(hdfs dfs -cat /out/JoinMapSide/part-* | sort) > /dev/null && echo "identical (459,636 rows each, compared after sorting)"
echo "== map task log of JoinMapSide: the cached menu"
app=$(grep -o "application_[0-9_]*" /tmp/JoinMapSide.log | head -1); sleep 5
yarn logs -applicationId $app -log_files stderr 2>/dev/null | grep -E "cache files|menu loaded" | head -2 | cut -c1-200

# Inside the namenode container:  bash /tmp/mr9/join_lab.sh

setup() once, map() many times

Watch the dictionary being built once, then used for every order. Coffee is not in the menu.

Worked examples: real problems, solved and traced in memory

Three map-side join decisions: memory sizing, choosing the join, and how much the cache really copies.

20,000 × 60 B · 5× overhead · 300 MB heap

raw 1.14 MB -> about 5.7 MB of heap; fits: True

3 pairs of tables

a -> replicated map-side join (distributed cache)
b -> composite map-side join (CompositeInputFormat)
c -> reduce-side join

400 tasks · 50 nodes · 30 MB

distributed cache: 1,500 MB   naive per task: 12,000 MB

Check yourself

Quick revision

Interview and viva questions

Practice problems

Map-side join exercises using the real numbers from the lab.

459,636 + 40,364

True

print(459636 + 40364 == 500000)

9,020,852 + 8,993,924 + 2,836,286

20851062 vs 20851062

print(9020852 + 8993924 + 2836286, "vs", 20851062)

24,828 ms vs 18,943 ms

5.9 s faster, 24% less time

print(f"{(24828 - 18943) / 1000:.1f} s faster, {1 - 18943 / 24828:.0%} less time")

menu + 2 orders

7,S0010,2026-09-02 13:05,thali,80 -> meal 80
8,S0011,2026-09-02 13:06,chai,15 -> drink 15

menu = ["chai,drink,15,A", "thali,meal,80,D"]
lookup = {}
for line in menu:
    item, cat, price, counter = line.split(",")
    lookup[item] = (cat, int(price))
for o in ["7,S0010,2026-09-02 13:05,thali,80", "8,S0011,2026-09-02 13:06,chai,15"]:
    cat, price = lookup[o.split(",")[3]]
    print(o, "->", cat, price)

3 orders

1 mismatched: ['2,S2,t,dosa,40']

price = {"chai": 15, "dosa": 45}
orders = ["1,S1,t,chai,15", "2,S2,t,dosa,40", "3,S3,t,dosa,45"]
bad = [o for o in orders if int(o.split(",")[4]) != price[o.split(",")[3]]]
print(len(bad), "mismatched:", bad)

3 tasks

tasks on 3 node(s): 3 download(s)
tasks on 1 node(s): 1 download(s)

for nodes_used in (3, 1):
    print(f"tasks on {nodes_used} node(s): {nodes_used} download(s)")

3 TB · 1 GB/s · 5 MB × 400 nodes

shuffle ~51 min of network time; broadcast 2.0 GB in total

shuffle_s = 3 * 1024 / 1
broadcast_mb = 5 * 400
print(f"shuffle ~{shuffle_s / 60:.0f} min of network time; broadcast {broadcast_mb / 1024:.1f} GB in total")

2 × 2 sorted partitions

1 dosa UPI
3 chai UPI
4 idli card
6 thali UPI

A = [[(1, "dosa"), (3, "chai")], [(4, "idli"), (6, "thali")]]
B = [[(1, "UPI"), (2, "cash"), (3, "UPI")], [(4, "card"), (6, "UPI")]]
for pa, pb in zip(A, B):
    i = j = 0
    while i < len(pa) and j < len(pb):
        if pa[i][0] == pb[j][0]:
            print(pa[i][0], pa[i][1], pb[j][1]); i += 1; j += 1
        elif pa[i][0] < pb[j][0]:
            i += 1
        else:
            j += 1

Real-world scenario questions (asked in company interviews)

Join strategies are a favourite interview topic for data engineers, from MapReduce to Spark.