The HDFS Java API

When the shell is not enough

The Hadoop FileSystem API

Words you'll meet in this lesson

API facts worth remembering

The API at three levels

API methods, shell commands and what happens underneath

Ways to reach HDFS from a program

Two real programs on the lab cluster

Both Java files are in “Run it on your PC” below. We compiled and ran them inside the namenode container, where Java and the Hadoop jars are already installed.

1. Compile and package

   javac -cp $(hadoop classpath) HdfsBasics.java ItemCount.java
   jar cf hdfs-demo.jar *.class
   hadoop jar hdfs-demo.jar HdfsBasics

   Connected to: hdfs://namenode
   hadoop classpath lists every Hadoop jar. hadoop jar runs the class with that classpath and the cluster's configuration, so new Configuration() finds fs.defaultFS=hdfs://namenode and FileSystem.get() returns a DistributedFileSystem.

2. Write and read back

   try (FSDataOutputStream out = fs.create(file, true)) {
       out.write("order_id,item,amount\n".getBytes(StandardCharsets.UTF_8));
       out.write("3001,samosa,18\n3002,chai,15\n".getBytes(StandardCharsets.UTF_8));
   }
   try (FSDataInputStream in = fs.open(file); BufferedReader r = ...) {
       while ((line = r.readLine()) != null) System.out.println("  read: " + line);
   }

   read: order_id,item,amount read: 3001,samosa,18 read: 3002,chai,15
   create = NameNode create + addBlock; the bytes went through a 3-DataNode pipeline; closing the stream completed the file. open fetched block locations and streamed from a DataNode.

3. Metadata and block locations

   FileStatus st = fs.getFileStatus(file);
   for (BlockLocation b : fs.getFileBlockLocations(bs, 0, bs.getLen())) ...

   orders.csv 49 bytes replication 3 block size 128 MB owner hadoop block at offset 0 length 134,217,728 on ab3b1bedf98a, 46201bb62e68, 264eb9c0f8e2 block at offset 134,217,728 length 6,542,542 on ab3b1bedf98a, 46201bb62e68, 264eb9c0f8e2
   The hosts are Docker container IDs: datanode-1, datanode-2 and datanode-3. orders_x30.csv (140,760,270 bytes) has a full 128 MB block plus a 6.5 MB tail block. A job scheduler uses exactly this list to run tasks next to the data.

4. List, rename, delete

   System.out.println("rename: " + fs.rename(file, renamed));
   System.out.println("exists after rename: " + fs.exists(file));
   System.out.println("delete: " + fs.delete(dir, true));

   ls: hdfs://namenode/user/hadoop/api_demo/orders.csv rename: true exists after rename: false delete: true
   Both return booleans. A rename to a missing parent folder returns false without an exception, so check it. delete(dir, true) removed the folder and the file permanently: the API skips the trash.

5. Count items in one file

   hadoop jar hdfs-demo.jar ItemCount /user/hadoop/data/canteen_orders.csv

   chai 50,170 dosa 49,948 idli 49,984 poha 49,891 samosa 50,021 vada pav 49,986 took 1327 ms in one JVM
   300,000 orders (4.5 MB) streamed through one client in 1.3 seconds, including connecting. The program reads the file block after block like -cat does.

6. Same program, 90× the data, and a bug

   hadoop jar hdfs-demo.jar ItemCount /user/hadoop/data/orders_x90.csv

   chai 4,515,300 dosa 4,495,320 idli 4,498,560 item 89 poha 4,490,190 samosa 4,501,890 vada pav 4,498,740 took 5890 ms in one JVM
   403 MB took 5.9 s: about 68 MB/s through one client. A 4 TB file would take about 17 hours this way, which is why MapReduce (Module 8) runs one reader per block in parallel. And “item 89”: orders_x90.csv was made by concatenating the file 90 times, so 89 header lines sit inside the data. HDFS stores bytes; cleaning data is the program's job.

7. The same file over HTTP: WebHDFS

   curl -s "http://namenode:9870/webhdfs/v1/user/hadoop/canteen/raw/day1.csv?op=GETFILESTATUS&user.name=hadoop"
   curl -s -i "http://namenode:9870/webhdfs/v1/user/hadoop/canteen/raw/day1.csv?op=OPEN&user.name=hadoop" | grep -iE "^HTTP|^Location"

   {"FileStatus":{"accessTime":1792579946689,"blockSize":134217728,"childrenNum":0,"fileId":16425,"group":"supergroup","length":47,"modificationTime":1792575484677,"owner":"hadoop","pathSuffix":"","permission":"640","replication":3,"storagePolicy":0,"type":"FILE"}} HTTP/1.1 307 Temporary Redirect Location: http://46201bb62e68:9864/webhdfs/v1/user/hadoop/canteen/raw/day1.csv?op=OPEN&user.name=hadoop&namenoderpcaddress=namenode:8020&offset=0
   The NameNode answers metadata itself but never sends file data: for OPEN it redirects (HTTP 307) to a DataNode (datanode-2, port 9864). The same split as the Java client, in plain HTTP.

Six API calls, and who answers each

The calls from HdfsBasics, in order.

The API at work

Simplified; organisations differ in tools and policies.

The FileSystem API, modelled in Python

The Java API can't run in your browser, so this Python class copies its shape: create, open, get_file_status, get_file_block_locations, rename and delete, with a NameNode namespace and 128 MB blocks placed on DataNodes. Run it, then try the tasks.

BLOCK = 128 * 1024 * 1024

class FileSystem:
    def __init__(self, datanodes):
        self.ns = {}                       # path -> (length, replication, blocks)
        self.dns = datanodes
        self.next_dn = 0

    def create(self, path, data_len, replication=3):
        blocks = []
        for off in range(0, data_len, BLOCK):
            hosts = [self.dns[(self.next_dn + i) % len(self.dns)] for i in range(replication)]
            self.next_dn += 1
            blocks.append((off, min(BLOCK, data_len - off), hosts))
        self.ns[path] = (data_len, replication, blocks)

    def get_file_status(self, path):
        length, rep, blocks = self.ns[path]
        return {"path": path, "len": length, "replication": rep, "blocksize": BLOCK}

    def get_file_block_locations(self, path):
        return self.ns[path][2]

    def rename(self, src, dst):            # returns False instead of raising, like Java
        if src not in self.ns or dst in self.ns:
            return False
        self.ns[dst] = self.ns.pop(src)
        return True

    def delete(self, path):
        return self.ns.pop(path, None) is not None

fs = FileSystem(["datanode-1", "datanode-2", "datanode-3"])
fs.create("/user/hadoop/data/orders_x30.csv", 140_760_270)
print(fs.get_file_status("/user/hadoop/data/orders_x30.csv"))
for off, length, hosts in fs.get_file_block_locations("/user/hadoop/data/orders_x30.csv"):
    print(f"  block at offset {off:,} length {length:,} on {', '.join(hosts)}")
print("rename:", fs.rename("/user/hadoop/data/orders_x30.csv", "/user/hadoop/data/old.csv"))
print("rename again:", fs.rename("/user/hadoop/data/orders_x30.csv", "/user/hadoop/data/old.csv"))

• for off in range(0, data_len, BLOCK):The file is cut into 128 MB blocks; the last one is shorter. The real file gave 134,217,728 + 6,542,542 bytes.
• def rename(self, src, dst): # returns False instead of raising, like JavaJava's fs.rename() also returns a boolean. The real program printed rename: true. Code that ignores a false silently loses track of files.
• def get_file_block_locations(self, path):Only metadata: the NameNode knows the hosts of every block. Schedulers use this to run tasks next to the data.

Run it on your PC: Java against HDFS

Save these files in the hdfs-lab folder. The namenode container already has Java and the Hadoop jars, so we copy the files in, compile there with javac -cp $(hadoop classpath), package a jar and run it with hadoop jar. The exact commands are at the end of the first file.

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

public class HdfsBasics {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();          // reads core-site.xml, hdfs-site.xml
        try (FileSystem fs = FileSystem.get(conf)) {
            System.out.println("Connected to: " + fs.getUri());

            Path dir = new Path("/user/hadoop/api_demo");
            fs.mkdirs(dir);

            // write a small file
            Path file = new Path(dir, "orders.csv");
            try (FSDataOutputStream out = fs.create(file, true)) {   // true = overwrite
                out.write("order_id,item,amount\n".getBytes(StandardCharsets.UTF_8));
                out.write("3001,samosa,18\n3002,chai,15\n".getBytes(StandardCharsets.UTF_8));
            }

            // read it back
            try (FSDataInputStream in = fs.open(file);
                 BufferedReader r = new BufferedReader(new InputStreamReader(in, StandardCharsets.UTF_8))) {
                String line;
                while ((line = r.readLine()) != null) {
                    System.out.println("  read: " + line);
                }
            }

            // metadata, as the NameNode stores it
            FileStatus st = fs.getFileStatus(file);
            System.out.printf("%s  %d bytes  replication %d  block size %d MB  owner %s%n",
                    st.getPath().getName(), st.getLen(), st.getReplication(),
                    st.getBlockSize() / (1024 * 1024), st.getOwner());

            // where are the blocks of a big file?
            Path big = new Path("/user/hadoop/data/orders_x30.csv");
            FileStatus bs = fs.getFileStatus(big);
            for (BlockLocation b : fs.getFileBlockLocations(bs, 0, bs.getLen())) {
                System.out.printf("  block at offset %,d length %,d on %s%n",
                        b.getOffset(), b.getLength(), String.join(", ", b.getHosts()));
            }

            // list, rename, delete
            for (FileStatus s : fs.listStatus(dir)) {
                System.out.println("  ls: " + s.getPath());
            }
            Path renamed = new Path(dir, "orders_old.csv");
            System.out.println("rename: " + fs.rename(file, renamed));
            System.out.println("exists after rename: " + fs.exists(file));
            System.out.println("delete: " + fs.delete(dir, true));             // true = recursive
        }
    }
}

// In the hdfs-lab folder:
//   docker cp HdfsBasics.java lab5-namenode-1:/tmp/
//   docker cp ItemCount.java lab5-namenode-1:/tmp/
//   docker exec -it lab5-namenode-1 bash
// Then inside the container:
//   cd /tmp
//   javac -cp $(hadoop classpath) HdfsBasics.java ItemCount.java
//   jar cf hdfs-demo.jar *.class
//   hadoop jar hdfs-demo.jar HdfsBasics

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.TreeMap;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

/** Count orders per item by streaming one HDFS file through a single client. */
public class ItemCount {
    public static void main(String[] args) throws Exception {
        Path path = new Path(args[0]);
        Map<String, Integer> counts = new TreeMap<>();
        long start = System.currentTimeMillis();
        try (FileSystem fs = FileSystem.get(new Configuration());
             FSDataInputStream in = fs.open(path);
             BufferedReader r = new BufferedReader(new InputStreamReader(in, StandardCharsets.UTF_8))) {
            r.readLine();                                         // skip the header
            String line;
            while ((line = r.readLine()) != null) {
                String item = line.split(",")[1];
                counts.merge(item, 1, Integer::sum);
            }
        }
        counts.forEach((item, n) -> System.out.printf("%-9s %,d%n", item, n));
        System.out.println("took " + (System.currentTimeMillis() - start) + " ms in one JVM");
    }
}

// Inside the namenode container, in /tmp:
//   hadoop jar hdfs-demo.jar ItemCount /user/hadoop/data/canteen_orders.csv
//   hadoop jar hdfs-demo.jar ItemCount /user/hadoop/data/orders_x90.csv

#!/usr/bin/env bash
# webhdfs.sh: the same file system over plain HTTP (REST). No Java, no Hadoop client needed.
# Run inside the namenode container:  bash webhdfs.sh
NN=http://namenode:9870/webhdfs/v1
F=/user/hadoop/canteen/raw/day1.csv
echo "--- GETFILESTATUS (metadata, answered by the NameNode)"
curl -s "$NN$F?op=GETFILESTATUS&user.name=hadoop"; echo
echo "--- OPEN without following redirects"
curl -s -i "$NN$F?op=OPEN&user.name=hadoop" | grep -iE "^HTTP|^Location"
echo "--- OPEN, following the redirect to the DataNode (-L)"
curl -s -L "$NN$F?op=OPEN&user.name=hadoop"

# In the hdfs-lab folder:
#   docker cp webhdfs.sh lab5-namenode-1:/tmp/
#   docker exec lab5-namenode-1 bash /tmp/webhdfs.sh

Reading a file block by block, as open() does

A simplified reader: ask for block locations once, then read each block from its first (nearest) host until the file's length is reached.

Worked examples: real problems, solved and traced in memory

Three API questions with numbers: what one JVM saves over shell calls, how fast one client can scan, and how many readers you need. Each runs live and is traced in memory.

2,000 calls · 1.8 s vs 0.02 s

shell 60 min, API 40 s, saved 59 min per day

402.7 MB in 5.89 s

1 TB: 4.3 hours at 68 MB/s
4 TB: 17.0 hours at 68 MB/s

4 TB · 128 MB blocks · 68 MB/s · 1 hour

32,768 blocks, 18 readers, about 1821 blocks each

Check yourself

Quick revision

Interview and viva questions

Practice problems

API exercises using the real outputs from the lab programs.

"order_id,item,amount\n3001,samosa,18\n3002,chai,15\n"

49 bytes

data = "order_id,item,amount\n" + "3001,samosa,18\n3002,chai,15\n"
print(len(data.encode("utf-8")), "bytes")

50,170 · 49,948 · 49,984 · 49,891 · 50,021 · 49,986

300000

print(sum([50170, 49948, 49984, 49891, 50021, 49986]))

140,760,270 bytes

[134217728, 6542542]

size, block = 140_760_270, 128 * 1024 * 1024
print([min(block, size - off) for off in range(0, size, block)])

x1 and x90 counts

True header lines inside: 89

x1 = {"chai": 50170, "dosa": 49948, "idli": 49984, "poha": 49891, "samosa": 50021, "vada pav": 49986}
x90 = {"chai": 4515300, "dosa": 4495320, "idli": 4498560, "poha": 4490190, "samosa": 4501890, "vada pav": 4498740}
print(all(x90[k] == 90 * x1[k] for k in x1), "header lines inside:", 90 - 1)

{"FileStatus":{…}}

47 3 640 128 MB

import json
reply = '{"FileStatus":{"accessTime":1792579946689,"blockSize":134217728,"childrenNum":0,"fileId":16425,"group":"supergroup","length":47,"modificationTime":1792575484677,"owner":"hadoop","pathSuffix":"","permission":"640","replication":3,"storagePolicy":0,"type":"FILE"}}'
st = json.loads(reply)["FileStatus"]
print(st["length"], st["replication"], st["permission"], st["blockSize"] // 2**20, "MB")

Location: http://46201bb62e68:9864/…

46201bb62e68 9864 0 namenode:8020

from urllib.parse import urlparse, parse_qs
loc = "http://46201bb62e68:9864/webhdfs/v1/user/hadoop/canteen/raw/day1.csv?op=OPEN&user.name=hadoop&namenoderpcaddress=namenode:8020&offset=0"
u = urlparse(loc); q = parse_qs(u.query)
print(u.hostname, u.port, q["offset"][0], q["namenoderpcaddress"][0])

60 vs 2 min/day · Rs 12/vCPU-hour

Rs 4,234 per year

rate = 12
saved_hours = (60 - 2) / 60 * 365
print(f"Rs {saved_hours * rate:,.0f} per year")

6 blocks · hosts per block

{'dn-1': [(0, 'local'), (4, 'local')], 'dn-2': [(1, 'local'), (3, 'local')], 'dn-3': [(2, 'local'), (5, 'local')]}
local: 6 of 6

locs = {0: ["dn-1", "dn-2", "dn-3"], 1: ["dn-1", "dn-2", "dn-3"], 2: ["dn-1", "dn-3", "dn-4"],
        3: ["dn-2", "dn-4", "dn-5"], 4: ["dn-1", "dn-5", "dn-2"], 5: ["dn-4", "dn-5", "dn-3"]}
workers = {"dn-1": [], "dn-2": [], "dn-3": []}
for b, hosts in locs.items():
    local = [w for w in workers if w in hosts]
    pick = min(local or workers, key=lambda w: len(workers[w]))
    workers[pick].append((b, "local" if pick in hosts else "remote"))
print(workers)
print("local:", sum(t == "local" for v in workers.values() for _, t in v), "of", len(locs))

Real-world scenario questions (asked in company interviews)

API questions check whether you can go beyond the shell and understand what client code really does.