Hadoop Streaming with Python

Riya writes MapReduce without Java

Hadoop Streaming

Words you'll meet in this lesson

From our lab (Hadoop 3.5.0, HDFS + YARN, 3 workers)

Streaming at three levels

Streaming options you will actually use

Streaming vs Java MapReduce vs PySpark

Revenue per category in Python, on the cluster

Run inside the namenode container of the Module 8 lab, with student_orders.csv (Lesson 9.1) in HDFS. The full script is streaming_lab.sh in “Run it on your PC”.

1. The mapper: one order in, one line out

   menu = {}
   with open("menu.csv") as f:                 # shipped with -files
       for line in f:
           item, category, price, counter = line.strip().split(",")
           menu[item] = category
   
   for line in sys.stdin:
       order_id, student, time, item, amount = line.strip().split(",")
       print(f"{menu.get(item, 'unknown')}\t{amount}")

   The menu lookup makes this a map-side join (Lesson 9.6) for free: menu.csv lands in the task's working folder because of -files.

2. The reducer: watch for the key to change

   current, orders, total = None, 0, 0
   for line in sys.stdin:
       category, amount = line.rstrip("\n").split("\t")
       if category != current:
           if current is not None:
               print(f"{current}\t{orders}\t{total}")
           current, orders, total = category, 0, 0
       orders += 1
       total += int(amount)
   if current is not None:
       print(f"{current}\t{orders}\t{total}")

   Two easy-to-miss lines: the final print after the loop (otherwise the last category is lost) and the if current is not None guard (an empty partition must print nothing).

3. Test with a Linux pipe first

   head -1000 student_orders.csv | python3 revenue_mapper.py | sort | python3 revenue_reducer.py

   breakfast 179 4940 drink 377 6060 meal 162 9145 snack 282 5362
   179 + 377 + 162 + 282 = 1,000 orders. sort stands in for the shuffle. If this works, the cluster run almost always works too.

4. Run it on the cluster

   mapred streaming -files revenue_mapper.py,revenue_reducer.py,menu.csv \
     -input /user/hadoop/data/student_orders.csv -output /out/stream \
     -mapper "python3 revenue_mapper.py" -reducer "python3 revenue_reducer.py"

   took 24512 ms number of splits:3 Job job_1792643020992_0045 completed successfully Launched map tasks=3 Map output records=500000 Reduce input groups=4 CPU time spent (ms)=5710
   Three map tasks each started a python3 process; the same counters as a Java job appear because the Java task around your script still does all the bookkeeping.

5. Read the output and cross-check

   hdfs dfs -cat /out/stream/part-00000

   breakfast 90110 2476620 drink 190033 3052315 meal 80219 4670075 snack 139638 2652992
   Streaming output files are called part-00000 (the older API's naming). Every number equals the Java counters job of Lesson 9.4: BREAKFAST 90,110 and ₹24,76,620, and so on.

6. Break the mapper and find the error

   # a typo: itme instead of item
   mapred streaming -files broken_mapper.py,revenue_reducer.py,menu.csv … -mapper "python3 broken_mapper.py" …
   yarn logs -applicationId <app id> -log_files stderr

   Error: java.lang.RuntimeException: PipeMapRed.waitOutputThreads(): subprocess failed with code 1 Job job_1792643020992_0046 failed with state FAILED due to: Task failed task_1792643020992_0046_m_000002 -- the Python error is in the task's stderr: NameError: name 'itme' is not defined Traceback (most recent call last):
   The job log only says that the subprocess exited with code 1. Python's traceback went to the task's stderr, which YARN keeps in the container logs. Retrying cannot fix a typo: the failed attempts are retried (up to 4 by default) and fail the same way, so the job gives up.

Lines through the pipes

Six steps of one Streaming job: ship the files, map through a Python process, sort, reduce through another Python process, and what a failure looks like.

Streaming at work

Simplified; organisations differ in tools and policies.

Simulate a Streaming job: mapper, sort, reducer

The mapper and reducer below are the lab scripts turned into functions that take and return lines. sorted() plays the shuffle, exactly as sort does in the local pipe test.

menu = {"chai": "drink", "coffee": "drink", "samosa": "snack", "vada pav": "snack",
        "poha": "breakfast", "idli": "breakfast", "dosa": "meal", "thali": "meal"}
orders = ["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,samosa,18", "4,S0001,2026-09-02 08:10,chai,15",
          "5,S0002,2026-09-02 13:05,thali,80", "6,S0001,2026-09-03 08:20,poha,25"]

def mapper(lines):                      # stdin lines in, "key\tvalue" lines out
    out = []
    for line in lines:
        oid, student, time, item, amount = line.split(",")
        out.append(f"{menu.get(item, 'unknown')}\t{amount}")
    return out

def reducer(lines):                     # SORTED lines in; no list of values per key
    out, current, orders, total = [], None, 0, 0
    for line in lines:
        key, value = line.split("\t")
        if key != current:
            if current is not None:
                out.append(f"{current}\t{orders}\t{total}")
            current, orders, total = key, 0, 0
        orders += 1
        total += int(value)
    if current is not None:
        out.append(f"{current}\t{orders}\t{total}")
    return out

mapped = mapper(orders)
print("map output :", mapped)
shuffled = sorted(mapped)               # the shuffle and sort
print("sorted     :", shuffled)
for line in reducer(shuffled):
    print("reduce     :", line)

• out.append(f"{menu.get(item, 'unknown')}\t{amount}")Everything is text. The tab separates the key (category) from the value (amount), exactly what Hadoop splits on.
• if key != current:The reducer only ever sees one line at a time, so it must notice where one key ends and the next begins.
• if current is not None:After the loop the last key is still open: forget this and the last category silently disappears.

Run it on your PC: Streaming on the Module 8 lab

Use the Module 8 lab cluster (Lesson 8.1). Copy the two scripts, menu.csv and student_orders.csv (from make_student_orders.py, Lesson 9.1) into /tmp/mr9 in the namenode container. The workers already have python3, because the apache/hadoop image includes it; on a real cluster check every NodeManager.

#!/usr/bin/env python3
# revenue_mapper.py: Hadoop Streaming mapper. Reads order lines on stdin, writes "category<TAB>amount".
import sys

menu = {}
with open("menu.csv") as f:                 # shipped with -files, appears in the task's working folder
    for line in f:
        item, category, price, counter = line.strip().split(",")
        menu[item] = category

for line in sys.stdin:
    order_id, student, time, item, amount = line.strip().split(",")
    print(f"{menu.get(item, 'unknown')}\t{amount}")

# Test alone:  head -3 student_orders.csv | python3 revenue_mapper.py

#!/usr/bin/env python3
# revenue_reducer.py: Hadoop Streaming reducer. Lines arrive SORTED by key; we detect where a key changes.
import sys

current, orders, total = None, 0, 0
for line in sys.stdin:
    category, amount = line.rstrip("\n").split("\t")
    if category != current:
        if current is not None:
            print(f"{current}\t{orders}\t{total}")
        current, orders, total = category, 0, 0
    orders += 1
    total += int(amount)
if current is not None:
    print(f"{current}\t{orders}\t{total}")

# Test the whole job locally:
#   head -1000 student_orders.csv | python3 revenue_mapper.py | sort | python3 revenue_reducer.py

# streaming_lab.sh: the same category-revenue job written in Python and run with Hadoop Streaming
cd /tmp/mr9
JAR=$(ls /opt/hadoop/share/hadoop/tools/lib/hadoop-streaming-*.jar); echo "streaming jar: $JAR"
O=/user/hadoop/data/student_orders.csv
echo "== 1. test the scripts locally first, with a Linux pipe (sort = the shuffle)"
head -1000 student_orders.csv | python3 revenue_mapper.py | sort | python3 revenue_reducer.py
echo "== 2. run on the cluster"
hdfs dfs -rm -r -f -skipTrash /out/stream > /dev/null 2>&1
start=$(date +%s%N)
mapred streaming -files revenue_mapper.py,revenue_reducer.py,menu.csv \
  -input $O -output /out/stream \
  -mapper "python3 revenue_mapper.py" -reducer "python3 revenue_reducer.py" > /tmp/stream.log 2>&1
echo "took $(( ($(date +%s%N) - start) / 1000000 )) ms"
grep -E "number of splits|completed successfully|Launched map tasks|Map output records|Reduce input groups|CPU time spent" /tmp/stream.log | sed -E 's/^[0-9-]+ [0-9:]+ INFO  //; s/^\s*/  /'
hdfs dfs -cat /out/stream/part-00000
echo "== 3. a bug in the mapper (a typo in a variable name)"
sed 's/print(f"{menu.get(item/print(f"{menu.get(itme/' revenue_mapper.py > broken_mapper.py
hdfs dfs -rm -r -f -skipTrash /out/stream_bad > /dev/null 2>&1
mapred streaming -files broken_mapper.py,revenue_reducer.py,menu.csv -input $O -output /out/stream_bad \
  -mapper "python3 broken_mapper.py" -reducer "python3 revenue_reducer.py" > /tmp/stream_bad.log 2>&1
grep -E "subprocess failed|failed with state" /tmp/stream_bad.log | sed -E 's/^[0-9-]+ [0-9:]+ (INFO|ERROR)  //' | sort -u | head -3 | cut -c1-160
app=$(grep -o "application_[0-9_]*" /tmp/stream_bad.log | head -1); sleep 5
echo "-- the Python error is in the task's stderr:"
yarn logs -applicationId $app -log_files stderr 2>/dev/null | grep -E "NameError|Traceback" | sort -u | head -2

# In the mr-lab folder:
#   docker cp revenue_mapper.py revenue_reducer.py menu.csv streaming_lab.sh lab8-namenode-1:/tmp/mr9/
#   docker exec lab8-namenode-1 bash /tmp/mr9/streaming_lab.sh

The key-change reducer, line by line

Four sorted lines with two keys. Watch current and total: the first key is printed only when the second key arrives, the last one only after the loop.

Worked examples: real problems, solved and traced in memory

Three Streaming problems: designing the scripts, a map-only job, and estimating the cost of text pipes.

5 order lines

{'08': 3, '13': 2}

-numReduceTasks 0 · 3 splits

2 matching lines: ['2,S2,t,thali,80', '3,S3,t,thali,80']
output files with 3 splits: ['part-00000', 'part-00001', 'part-00002']

2 µs per crossing · 3 crossings per record

    500,000 records -> 3 s of extra CPU
500,000,000 records -> 3,000 s of extra CPU

Check yourself

Quick revision

Interview and viva questions

Practice problems

Streaming exercises with the real output of our lab run.

4 counts

1000

print(179 + 377 + 162 + 282)

3 lines

'meal' '45'
'drink' '15\textra'
'notab' ''

for line in ["meal\t45", "drink\t15\textra", "notab"]:
    key, _, value = line.partition("\t")
    print(repr(key), repr(value))

4 output lines

breakfast Rs 27.48
drink     Rs 16.06
meal      Rs 58.22
snack     Rs 19.00

out = {"breakfast": (90110, 2476620), "drink": (190033, 3052315), "meal": (80219, 4670075), "snack": (139638, 2652992)}
for c, (n, total) in out.items():
    print(f"{c:9} Rs {total / n:.2f}")

sorted lines

drink 3 15 20
meal 2 45 80

lines = sorted(["drink\t15", "drink\t20", "meal\t45", "meal\t80", "drink\t15"])
current, vals = None, []
for line in lines + [None]:                     # None marks the end of input
    key, value = line.split("\t") if line else (None, None)
    if key != current and current is not None:
        print(current, len(vals), min(vals), max(vals))
        vals = []
    current = key
    if value is not None:
        vals.append(int(value))

3 lines

reporter:counter:Quality,BAD_LINES,1
reporter:counter:Quality,BAD_LINES,1

lines = ["1,S1,t,chai,15", "broken line", "3,S3,t,dosa"]
for line in lines:
    if len(line.split(",")) != 5:
        print("reporter:counter:Quality,BAD_LINES,1")   # in a real script: sys.stderr.write(... + "\n")

combiner output fed to the reducer

reducer would read value: '3\t50' -> int() fails

combined = ["drink\t3\t50"]                         # what the reducer script would print as a combiner
key, value = combined[0].split("\t", 1)
print("reducer would read value:", repr(value), "-> int() fails")

40 → 12 minutes · 20 nodes · ₹50/node-hour

Rs 467 saved per night, pays back in 514 nights

saved_per_night = (40 - 12) / 60 * 20 * 50
print(f"Rs {saved_per_night:.0f} saved per night, pays back in {2 * 120000 / saved_per_night:.0f} nights")

5 lines, 2 reducers

reducer 0: S2 first order 2026-09-02 09:00 poha
reducer 1: S1 first order 2026-09-02 08:10 chai

lines = ["S2\t2026-09-03 08:05\tchai", "S1\t2026-09-02 13:05\tthali", "S1\t2026-09-02 08:10\tchai",
         "S2\t2026-09-02 09:00\tpoha", "S1\t2026-09-03 08:40\tpoha"]
parts = {0: [], 1: []}
for l in lines:
    parts[int(l.split("\t")[0][1:]) % 2].append(l)          # partition on field 1 only
for p in parts:
    current = None
    for l in sorted(parts[p]):                              # sort on fields 1 and 2
        s, t, item = l.split("\t")
        if s != current:
            print(f"reducer {p}: {s} first order {t} {item}")
            current = s

Real-world scenario questions (asked in company interviews)

Streaming comes up when interviewers check whether you know MapReduce beyond Java, and in debugging questions.