Course Outcomes: After completing this course, students will be able to

CO2: Apply appropriate tools and frameworks (Hadoop, Spark, NoSQL) for storing, processing,
and managing large-scale datasets
 3. The Hadoop Distributed Filesystem . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 41

4. Hadoop I/O . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 75

vi | Table of Contents
 CHAPTER 3
The Hadoop Distributed Filesystem

The Design of HDFS

41
HDFS Concepts
Blocks

42 | Chapter 3: The Hadoop Distributed Filesystem

Why Is a Block in HDFS So Large?

HDFS Concepts | 43
 fsck

% hadoop fsck -files -blocks

Namenodes and Datanodes

44 | Chapter 3: The Hadoop Distributed Filesystem

The Command-Line Interface

[Link]

hdfs

localhost

[Link]

Basic Filesystem Operations

hadoop fs -help

% hadoop fs -copyFromLocal input/docs/[Link] hdfs://localhost/user/tom/[Link]

The Command-Line Interface | 45

 fs
-copyFromLocal

hdfs://localhost
% hadoop fs -copyFromLocal input/docs/[Link] /user/tom/[Link]

% hadoop fs -copyFromLocal input/docs/[Link] [Link]

% hadoop fs -copyToLocal [Link] [Link]

% md5 input/docs/[Link] [Link]
MD5 (input/docs/[Link]) = a16f231da6b05e2ba7a339320e7dacd9
MD5 ([Link]) = a16f231da6b05e2ba7a339320e7dacd9

% hadoop fs -mkdir books

% hadoop fs -ls .
Found 2 items
drwxr-xr-x - tom supergroup 0 2009-04-02 22:41 /user/tom/books
-rw-r--r-- 1 tom supergroup 118 2009-04-02 22:29 /user/tom/[Link]

ls -l

46 | Chapter 3: The Hadoop Distributed Filesystem

 File Permissions in HDFS

r w
x

[Link]

Hadoop Filesystems
[Link]

Hadoop Filesystems | 47
 Filesystem URI scheme Java implementation (all under [Link]) Description
Local file [Link] A filesystem for a locally connec-
ted disk with client-side check-
sums. Use RawLocalFileSys
tem for a local filesystem with no
checksums. See “LocalFileSys-
tem” on page 76.
HDFS hdfs [Link] Hadoop’s distributed filesystem.
HDFS is designed to work effi-
ciently in conjunction with Map-
Reduce.
HFTP hftp [Link] A filesystem providing read-only
access to HDFS over HTTP. (Despite
its name, HFTP has no connection
with FTP.) Often used with distcp
(“Parallel Copying with
distcp” on page 70) to copy data
between HDFS clusters running
different versions.
HSFTP hsftp [Link] A filesystem providing read-only
access to HDFS over HTTPS. (Again,
this has no connection with FTP.)
HAR har [Link] A filesystem layered on another
filesystem for archiving files. Ha-
doop Archives are typically used
for archiving files in HDFS to reduce
the namenode’s memory usage.
See “Hadoop Ar-
chives” on page 71.
KFS (Cloud- kfs [Link] CloudStore (formerly Kosmos fil-
Store) esystem) is a distributed filesys-
tem like HDFS or Google’s GFS,
written in C++. Find more infor-
mation about it at [Link]
.[Link]/.
FTP ftp [Link] A filesystem backed by an FTP
server.
S3 (native) s3n fs.s3native.NativeS3FileSystem A filesystem backed by Amazon
S3. See [Link]
doop/AmazonS3.
S3 (block- s3 fs.s3.S3FileSystem A filesystem backed by Amazon
based) S3, which stores files in blocks
(much like HDFS) to overcome S3’s
5 GB file size limit.

48 | Chapter 3: The Hadoop Distributed Filesystem

 % hadoop fs -ls [Link]

Interfaces

FileSystem

Thrift

Writable

Hadoop Filesystems | 49
C
FileSystem

FUSE

ls cat

WebDAV

Other HDFS Interfaces

50 | Chapter 3: The Hadoop Distributed Filesystem

 HftpFile
System
HsftpFileSystem

FTPFileSystem

The Java Interface

FileSystem

DistributedFileSystem
FileSystem

Reading Data from a Hadoop URL

[Link]
InputStream in = null;
try {
in = new URL("hdfs://host/path").openStream();
// process in
} finally {
[Link](in);
}

hdfs
setURLStreamHandlerFactory URL
FsUrlStreamHandlerFactory

URLStreamHandlerFactory

FileSystem

The Java Interface | 51

 cat

public class URLCat {

static {
[Link](new FsUrlStreamHandlerFactory());
}

public static void main(String[] args) throws Exception {

InputStream in = null;
try {
in = new URL(args[0]).openStream();
[Link](in, [Link], 4096, false);
} finally {
[Link](in);
}
}
}

IOUtils
finally
[Link] copyBytes

[Link]

% hadoop URLCat hdfs://localhost/user/tom/[Link]

On the top of the Crumpetty Tree
The Quangle Wangle sat,
But his face you could not see,
On account of his Beaver Hat.

Reading Data Using the FileSystem API

URLStreamHand
lerFactory FileSystem

Path
[Link]
Path

52 | Chapter 3: The Hadoop Distributed Filesystem

FileSystem

FileSystem
public static FileSystem get(Configuration conf) throws IOException
public static FileSystem get(URI uri, Configuration conf) throws IOException

Configuration

URI

URI
FileSystem open()

public FSDataInputStream open(Path f) throws IOException

public abstract FSDataInputStream open(Path f, int bufferSize) throws IOException

public class FileSystemCat {

public static void main(String[] args) throws Exception {

String uri = args[0];
Configuration conf = new Configuration();
FileSystem fs = [Link]([Link](uri), conf);
InputStream in = null;
try {
in = [Link](new Path(uri));
[Link](in, [Link], 4096, false);
} finally {
[Link](in);
}
}
}

% hadoop FileSystemCat hdfs://localhost/user/tom/[Link]

On the top of the Crumpetty Tree
The Quangle Wangle sat,
But his face you could not see,
On account of his Beaver Hat.

The Java Interface | 53

FSDataInputStream
open() FileSystem FSDataInputStream
[Link] [Link]

package [Link];

public class FSDataInputStream extends DataInputStream

implements Seekable, PositionedReadable {
// implementation elided
}

Seekable
getPos()
public interface Seekable {
void seek(long pos) throws IOException;
long getPos() throws IOException;
boolean seekToNewSource(long targetPos) throws IOException;
}

seek()
IOException skip() [Link]
seek()

seekToNewSource()
targetPos

public class FileSystemDoubleCat {

public static void main(String[] args) throws Exception {

String uri = args[0];
Configuration conf = new Configuration();
FileSystem fs = [Link]([Link](uri), conf);
FSDataInputStream in = null;
try {
in = [Link](new Path(uri));
[Link](in, [Link], 4096, false);
[Link](0); // go back to the start of the file
[Link](in, [Link], 4096, false);
} finally {
[Link](in);
}

54 | Chapter 3: The Hadoop Distributed Filesystem

 }
}

% hadoop FileSystemDoubleCat hdfs://localhost/user/tom/[Link]

On the top of the Crumpetty Tree
The Quangle Wangle sat,
But his face you could not see,
On account of his Beaver Hat.
On the top of the Crumpetty Tree
The Quangle Wangle sat,
But his face you could not see,
On account of his Beaver Hat.

FSDataInputStream PositionedReadable

public interface PositionedReadable {

public int read(long position, byte[] buffer, int offset, int length)
throws IOException;

public void readFully(long position, byte[] buffer, int offset, int length)
throws IOException;

public void readFully(long position, byte[] buffer) throws IOException;

}

read() length position

buffer offset
length readFully()
length [Link]
buffer
EOFException

Seekable
long oldPos = getPos();
try {
seek(position);
// read data
} finally {
seek(oldPos);
}

seek()

The Java Interface | 55

Writing Data
FileSystem
Path

public FSDataOutputStream create(Path f) throws IOException

create()

exists()

Progressable

package [Link];

public interface Progressable {

public void progress();
}

append()
public FSDataOutputStream append(Path f) throws IOException

progress()

public class FileCopyWithProgress {

public static void main(String[] args) throws Exception {
String localSrc = args[0];
String dst = args[1];

56 | Chapter 3: The Hadoop Distributed Filesystem

 InputStream in = new BufferedInputStream(new FileInputStream(localSrc));

Configuration conf = new Configuration();

FileSystem fs = [Link]([Link](dst), conf);
OutputStream out = [Link](new Path(dst), new Progressable() {
public void progress() {
[Link](".");
}
});

[Link](in, out, 4096, true);

}
}

% hadoop FileCopyWithProgress input/docs/[Link] hdfs://localhost/user/tom/[Link]

...............

progress()

FSDataOutputStream
create() FileSystem FSDataOutputStream
FSDataInputStream
package [Link];

public class FSDataOutputStream extends DataOutputStream implements Syncable {

public long getPos() throws IOException {

// implementation elided
}

// implementation elided

FSDataInputStream FSDataOutputStream

Directories
FileSystem
public boolean mkdirs(Path f) throws IOException

[Link] mkdirs() true

The Java Interface | 57

create()

Querying the Filesystem

File metadata: FileStatus

FileStatus

getFileStatus() FileSystem FileStatus

public class ShowFileStatusTest {

private MiniDFSCluster cluster; // use an in-process HDFS cluster for testing

private FileSystem fs;

@Before
public void setUp() throws IOException {
Configuration conf = new Configuration();
if ([Link]("[Link]") == null) {
[Link]("[Link]", "/tmp");
}
cluster = new MiniDFSCluster(conf, 1, true, null);
fs = [Link]();
OutputStream out = [Link](new Path("/dir/file"));
[Link]("content".getBytes("UTF-8"));
[Link]();
}

@After
public void tearDown() throws IOException {
if (fs != null) { [Link](); }
if (cluster != null) { [Link](); }
}

@Test(expected = [Link])
public void throwsFileNotFoundForNonExistentFile() throws IOException {
[Link](new Path("no-such-file"));
}

@Test
public void fileStatusForFile() throws IOException {
Path file = new Path("/dir/file");
FileStatus stat = [Link](file);
assertThat([Link]().toUri().getPath(), is("/dir/file"));
assertThat([Link](), is(false));
assertThat([Link](), is(7L));

58 | Chapter 3: The Hadoop Distributed Filesystem

 assertThat([Link](),
is(lessThanOrEqualTo([Link]())));
assertThat([Link](), is((short) 1));
assertThat([Link](), is(64 * 1024 * 1024L));
assertThat([Link](), is("tom"));
assertThat([Link](), is("supergroup"));
assertThat([Link]().toString(), is("rw-r--r--"));
}

@Test
public void fileStatusForDirectory() throws IOException {
Path dir = new Path("/dir");
FileStatus stat = [Link](dir);
assertThat([Link]().toUri().getPath(), is("/dir"));
assertThat([Link](), is(true));
assertThat([Link](), is(0L));
assertThat([Link](),
is(lessThanOrEqualTo([Link]())));
assertThat([Link](), is((short) 0));
assertThat([Link](), is(0L));
assertThat([Link](), is("tom"));
assertThat([Link](), is("supergroup"));
assertThat([Link]().toString(), is("rwxr-xr-x"));
}

FileNotFoundException
exists()
FileSystem
public boolean exists(Path f) throws IOException

Listing files

FileSystem listStatus()

public FileStatus[] listStatus(Path f) throws IOException

public FileStatus[] listStatus(Path f, PathFilter filter) throws IOException
public FileStatus[] listStatus(Path[] files) throws IOException
public FileStatus[] listStatus(Path[] files, PathFilter filter) throws IOException

FileStatus
FileStatus

PathFilter

listStatus FileStatus

The Java Interface | 59

 stat2Paths() FileUtil FileStatus
Path

public class ListStatus {

public static void main(String[] args) throws Exception {

String uri = args[0];
Configuration conf = new Configuration();
FileSystem fs = [Link]([Link](uri), conf);

Path[] paths = new Path[[Link]];

for (int i = 0; i < [Link]; i++) {
paths[i] = new Path(args[i]);
}

FileStatus[] status = [Link](paths);

Path[] listedPaths = FileUtil.stat2Paths(status);
for (Path p : listedPaths) {
[Link](p);
}
}
}

% hadoop ListStatus hdfs://localhost/ hdfs://localhost/user/tom

hdfs://localhost/user
hdfs://localhost/user/tom/books
hdfs://localhost/user/tom/[Link]

File patterns

FileSystem
public FileStatus[] globStatus(Path pathPattern) throws IOException
public FileStatus[] globStatus(Path pathPattern, PathFilter filter) throws IOException

globStatus() FileStatus
PathFilter

60 | Chapter 3: The Hadoop Distributed Filesystem

 Glob Name Matches
* asterisk Matches zero or more characters
? question mark Matches a single character
[ab] character class Matches a single character in the set {a, b}
[^ab] negated character class Matches a single character that is not in the set {a, b}
[a-b] character range Matches a single character in the (closed) range [a, b], where a is lexicographically
less than or equal to b
[^a-b] negated character range Matches a single character that is not in the (closed) range [a, b], where a is
lexicographically less than or equal to b
{a,b} alternation Matches either expression a or b
\c escaped character Matches character c when it is a metacharacter

Glob Expansion
/* /2007 /2008
/*/* /2007/12 /2008/01
/*/12/* /2007/12/30 /2007/12/31
/200? /2007 /2008
/200[78] /2007 /2008
/200[7-8] /2007 /2008
/200[^01234569] /2007 /2008
/*/*/{31,01} /2007/12/31 /2008/01/01
/*/*/3{0,1} /2007/12/30 /2007/12/31
/*/{12/31,01/01} /2007/12/31 /2008/01/01

PathFilter

The Java Interface | 61

 listStatus() globStatus() FileSystem
PathFilter
package [Link];

public interface PathFilter {

boolean accept(Path path);
}

PathFilter [Link] Path File

PathFilter

public class RegexExcludePathFilter implements PathFilter {

private final String regex;

public RegexExcludePathFilter(String regex) {

[Link] = regex;
}

public boolean accept(Path path) {

return ![Link]().matches(regex);
}
}

[Link](new Path("/2007/*/*"), new RegexExcludeFilter("^.*/2007/12/31$"))

Path

PathFilter

Deleting Data
delete() FileSystem
public boolean delete(Path f, boolean recursive) throws IOException

f recursive
recursive true
IOException

62 | Chapter 3: The Hadoop Distributed Filesystem

Data Flow
Anatomy of a File Read

open() FileSystem
DistributedFileSystem
DistributedFileSystem

DistributedFileSystem FSDataInputStream
FSDataInputStream
DFSInputStream

Data Flow | 63
 read() DFSInputStream

read()
DFSInputStream

DFSInputStream

close() FSDataInputStream

Network Topology and Hadoop

64 | Chapter 3: The Hadoop Distributed Filesystem

Data Flow | 65
Anatomy of a File Write

create() DistributedFileSystem
DistributedFileSystem

IOException DistributedFileSystem FSDataOutputStream

FSDataOutputStream DFSOutput
Stream
DFSOutputStream
Data
Streamer

DataStreamer

66 | Chapter 3: The Hadoop Distributed Filesystem

DFSOutputStream

[Link]

[Link]
close()

Data
Streamer

Replica Placement

Data Flow | 67
Coherency Model

Path p = new Path("p");

[Link](p);
assertThat([Link](p), is(true));

Path p = new Path("p");

OutputStream out = [Link](p);

68 | Chapter 3: The Hadoop Distributed Filesystem

 [Link]("content".getBytes("UTF-8"));
[Link]();
assertThat([Link](p).getLen(), is(0L));

sync() FSDataOutputStream sync()

Path p = new Path("p");

FSDataOutputStream out = [Link](p);
[Link]("content".getBytes("UTF-8"));
[Link]();
[Link]();
assertThat([Link](p).getLen(), is(((long) "content".length())));

fsync

FileOutputStream out = new FileOutputStream(localFile);

[Link]("content".getBytes("UTF-8"));
[Link](); // flush to operating system
[Link]().sync(); // sync to disk
assertThat([Link](), is(((long) "content".length())));

sync()
Path p = new Path("p");
OutputStream out = [Link](p);
[Link]("content".getBytes("UTF-8"));
[Link]();
assertThat([Link](p).getLen(), is(((long) "content".length())));

Consequences for application design

sync()

sync()
sync()

sync()

Data Flow | 69
Parallel Copying with distcp

% hadoop distcp hdfs://namenode1/foo hdfs://namenode2/bar

-overwrite
-update

-overwrite -update

% hadoop distcp -update hdfs://namenode1/foo hdfs://namenode2/bar/foo

-overwrite -update

70 | Chapter 3: The Hadoop Distributed Filesystem

 -m -m 1000

% hadoop distcp h[Link] hdfs://namenode2/bar

[Link]

Keeping an HDFS Cluster Balanced

-m
1

Hadoop Archives

Hadoop Archives | 71
Using Hadoop Archives

% hadoop fs -lsr /my/files

-rw-r--r-- 1 tom supergroup 1 2009-04-09 19:13 /my/files/a
drwxr-xr-x - tom supergroup 0 2009-04-09 19:13 /my/files/dir
-rw-r--r-- 1 tom supergroup 1 2009-04-09 19:13 /my/files/dir/b

archive
% hadoop archive -archiveName [Link] /my/files /my

% hadoop fs -ls /my

Found 2 items
drwxr-xr-x - tom supergroup 0 2009-04-09 19:13 /my/files
drwxr-xr-x - tom supergroup 0 2009-04-09 19:13 /my/[Link]
% hadoop fs -ls /my/[Link]
Found 3 items
-rw-r--r-- 10 tom supergroup 165 2009-04-09 19:13 /my/[Link]/_index
-rw-r--r-- 10 tom supergroup 23 2009-04-09 19:13 /my/[Link]/_masterindex
-rw-r--r-- 1 tom supergroup 2 2009-04-09 19:13 /my/[Link]/part-0

% hadoop fs -lsr har:///my/[Link]

drw-r--r-- - tom supergroup 0 2009-04-09 19:13 /my/[Link]/my
drw-r--r-- - tom supergroup 0 2009-04-09 19:13 /my/[Link]/my/files

72 | Chapter 3: The Hadoop Distributed Filesystem

 -rw-r--r-- 10 tom supergroup 1 2009-04-09 19:13 /my/[Link]/my/files/a
drw-r--r-- - tom supergroup 0 2009-04-09 19:13 /my/[Link]/my/files/dir
-rw-r--r-- 10 tom supergroup 1 2009-04-09 19:13 /my/[Link]/my/files/dir/b

% hadoop fs -lsr har:///my/[Link]/my/files/dir

% hadoop fs -lsr har://hdfs-localhost:8020/my/[Link]/my/files/dir

% hadoop fs -rmr /my/[Link]

Limitations

InputFormat

Hadoop Archives | 73
 CHAPTER 4
Hadoop I/O

Data Integrity

Data Integrity in HDFS

[Link]

75
 ChecksumException IOException

DataBlockScanner

ChecksumException

false setVerify
Checksum() FileSystem open()
-ignoreCrc -get
-copyToLocal

LocalFileSystem
LocalFileSystem

[Link]

LocalFileSystem ChecksumException

76 | Chapter 4: Hadoop I/O

 RawLocalFileSystem
LocalFileSystem
[Link]
[Link] Raw
LocalFileSystem

Configuration conf = ...

FileSystem fs = new RawLocalFileSystem();
[Link](null, conf);

ChecksumFileSystem
LocalFileSystem ChecksumFileSystem
ChecksumFileSys
tem FileSystem
FileSystem rawFs = ...
FileSystem checksummedFs = new ChecksumFileSystem(rawFs);

getRawFileSystem() ChecksumFileSystem ChecksumFileSystem

getChecksumFile()

ChecksumFileSystem
reportChecksumFailure()
LocalFileSystem

Compression

Compression | 77
 Compression format Tool Algorithm Filename extension Multiple files Splittable
DEFLATEa N/A DEFLATE .deflate No No
gzip gzip DEFLATE .gz No No
ZIP zip DEFLATE .zip Yes Yes, at file boundaries
bzip2 bzip2 bzip2 .bz2 No Yes
LZO lzop LZO .lzo No No

–1 -9

gzip -1 file

78 | Chapter 4: Hadoop I/O

Codecs
CompressionCodec
GzipCodec

Compression format Hadoop CompressionCodec

DEFLATE [Link]
gzip [Link]
bzip2 [Link].BZip2Codec
LZO [Link]

LzopCodec lzop

LzoCodec

Compressing and decompressing streams with CompressionCodec

CompressionCodec
createOutput
Stream(OutputStream out) CompressionOutputStream

createInputStream(InputStream in) CompressionInputStream

CompressionOutputStream CompressionInputStream
[Link] [Link]

SequenceFile

public class StreamCompressor {

public static void main(String[] args) throws Exception {

Compression | 79
 String codecClassname = args[0];
Class<?> codecClass = [Link](codecClassname);
Configuration conf = new Configuration();
CompressionCodec codec = (CompressionCodec)
[Link](codecClass, conf);

CompressionOutputStream out = [Link]([Link]);

[Link]([Link], out, 4096, false);
[Link]();
}
}

CompressionCodec
ReflectionUtils
[Link]
copyBytes() IOUtils
CompressionOutputStream finish()
CompressionOutputStream

StreamCompressor
GzipCodec
% echo "Text" | hadoop StreamCompressor [Link] \
| gunzip -
Text

Inferring CompressionCodecs using CompressionCodecFactory

GzipCodec

CompressionCodecFactory
CompressionCodec getCodec() Path

public class FileDecompressor {

public static void main(String[] args) throws Exception {

String uri = args[0];
Configuration conf = new Configuration();
FileSystem fs = [Link]([Link](uri), conf);

Path inputPath = new Path(uri);

CompressionCodecFactory factory = new CompressionCodecFactory(conf);
CompressionCodec codec = [Link](inputPath);
if (codec == null) {
[Link]("No codec found for " + uri);
[Link](1);

80 | Chapter 4: Hadoop I/O

 }

String outputUri =
[Link](uri, [Link]());

InputStream in = null;
OutputStream out = null;
try {
in = [Link]([Link](inputPath));
out = [Link](new Path(outputUri));
[Link](in, out, conf);
} finally {
[Link](in);
[Link](out);
}
}
}

removeSuffix() CompressionCodecFactory

% hadoop FileDecompressor [Link]

CompressionCodecFactory
[Link]

CompressionCodecFactory

Property name Type Default value Description

[Link] comma-separated [Link]. A list of the Compres
Class names [Link], sionCodec classes for
[Link]. compression/
[Link], decompression.
[Link].
compress.Bzip2Codec

Native libraries

Compression | 81
 Compression format Java implementation Native implementation
DEFLATE Yes Yes
gzip Yes Yes
bzip2 Yes No
LZO No Yes

[Link]

[Link]
false
CodecPool.
CodecPool

Compressor

public class PooledStreamCompressor {

public static void main(String[] args) throws Exception {

String codecClassname = args[0];
Class<?> codecClass = [Link](codecClassname);
Configuration conf = new Configuration();
CompressionCodec codec = (CompressionCodec)
[Link](codecClass, conf);
Compressor compressor = null;
try {
compressor = [Link](codec);
CompressionOutputStream out =
[Link]([Link], compressor);
[Link]([Link], out, 4096, false);
[Link]();

82 | Chapter 4: Hadoop I/O

 } finally {
[Link](compressor);
}
}
}

Compressor CompressionCodec
createOutputStream() finally

IOException

Compression and Input Splits

Compression | 83
 Which Compression Format Should I Use?

Using Compression in MapReduce

[Link] true [Link]

public class MaxTemperatureWithCompression {

public static void main(String[] args) throws IOException {

if ([Link] != 2) {
[Link]("Usage: MaxTemperatureWithCompression <input path> " +
"<output path>");
[Link](-1);
}

JobConf conf = new JobConf([Link]);

84 | Chapter 4: Hadoop I/O

 [Link]("Max temperature with output compression");

[Link](conf, new Path(args[0]));

[Link](conf, new Path(args[1]));

[Link]([Link]);
[Link]([Link]);

[Link]("[Link]", true);
[Link]("[Link]", [Link],
[Link]);

[Link]([Link]);
[Link]([Link]);
[Link]([Link]);

[Link](conf);
}
}

% hadoop MaxTemperatureWithCompression input/ncdc/[Link] output

% gunzip -c output/[Link]
1949 111
1950 22

[Link]
[Link]
RECORD BLOCK

Compressing map output

Compression | 85
 Property name Type Default value Description
[Link] boolean false Compress map outputs.
[Link]. Class [Link]. The compression codec to use for
[Link] [Link] map outputs.

[Link](true);
[Link]([Link]);

Serialization

86 | Chapter 4: Hadoop I/O

The Writable Interface
DataOutput
DataInput
package [Link];

import [Link];
import [Link];
import [Link];

public interface Writable {

void write(DataOutput out) throws IOException;
void readFields(DataInput in) throws IOException;
}

Writable
IntWritable int
set()
IntWritable writable = new IntWritable();
[Link](163);

IntWritable writable = new IntWritable(163);

IntWritable
[Link] [Link]
[Link]
public static byte[] serialize(Writable writable) throws IOException {
ByteArrayOutputStream out = new ByteArrayOutputStream();
DataOutputStream dataOut = new DataOutputStream(out);
[Link](dataOut);
[Link]();

Serialization | 87
 return [Link]();
}

byte[] bytes = serialize(writable);

assertThat([Link], is(4));

[Link]
StringUtils
assertThat([Link](bytes), is("000000a3"));

Writable

public static byte[] deserialize(Writable writable, byte[] bytes)

throws IOException {
ByteArrayInputStream in = new ByteArrayInputStream(bytes);
DataInputStream dataIn = new DataInputStream(in);
[Link](dataIn);
[Link]();
return bytes;
}

IntWritable deserialize()

get()
IntWritable newWritable = new IntWritable();
deserialize(newWritable, bytes);
assertThat([Link](), is(163));

WritableComparable and comparators

IntWritable WritableComparable
Writable [Link]
package [Link];

public interface WritableComparable<T> extends Writable, Comparable<T> {

}

RawComparator Comparator
package [Link];

import [Link];

public interface RawComparator<T> extends Comparator<T> {

public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2);

88 | Chapter 4: Hadoop I/O

 }

IntWritable compare()
b1 b2
s1 s2 l1 l2
WritableComparator RawComparator
WritableComparable
compare()
compare()
RawComparator Writable
IntWritable
RawComparator<IntWritable> comparator = [Link]([Link]);

IntWritable
IntWritable w1 = new IntWritable(163);
IntWritable w2 = new IntWritable(67);
assertThat([Link](w1, w2), greaterThan(0));

byte[] b1 = serialize(w1);
byte[] b2 = serialize(w2);
assertThat([Link](b1, 0, [Link], b2, 0, [Link]),
greaterThan(0));

Writable Classes
Writable [Link]

Writable wrappers for Java primitives

Writable
short char IntWritable get()
set()

Java primitive Writable implementation Serialized size (bytes)

boolean BooleanWritable 1
byte ByteWritable 1
int IntWritable 4
VIntWritable 1–5
float FloatWritable 4

Serialization | 89
 Java primitive Writable implementation Serialized size (bytes)
long LongWritable 8
VLongWritable 1–9
double DoubleWritable 8

IntWritable LongWritable VIntWritable

VLongWritable

90 | Chapter 4: Hadoop I/O

 byte[] data = serialize(new VIntWritable(163));
assertThat([Link](data), is("8fa3"));

VIntWritable VLongWritable

long

Text
Text Writable Writable
[Link] Text UTF8

Text int
Text

Indexing.
Text String Text

char String
charAt()
Text t = new Text("hadoop");
assertThat([Link](), is(6));
assertThat([Link]().length, is(6));

assertThat([Link](2), is((int) 'd'));

assertThat("Out of bounds", [Link](100), is(-1));

charAt() int
String char Text find()
String indexOf()
Text t = new Text("hadoop");
assertThat("Find a substring", [Link]("do"), is(2));
assertThat("Finds first 'o'", [Link]("o"), is(3));
assertThat("Finds 'o' from position 4 or later", [Link]("o", 4), is(4));
assertThat("No match", [Link]("pig"), is(-1));

Serialization | 91
Unicode.
Text String

Unicode code point U+0041 U+00DF U+6771 U+10400

Name LATIN CAPITAL LATIN SMALL LET- N/A (a unified Han DESERET CAPITAL LETTER LONG I
LETTER A TER SHARP S ideograph)
UTF-8 code units 41 c3 9f e6 9d b1 f0 90 90 80
Java representation \u0041 \u00DF \u6771 \uuD801\uDC00

char char

String Text

public class StringTextComparisonTest {

@Test
public void string() throws UnsupportedEncodingException {

String s = "\u0041\u00DF\u6771\uD801\uDC00";
assertThat([Link](), is(5));
assertThat([Link]("UTF-8").length, is(10));

assertThat([Link]("\u0041"), is(0));
assertThat([Link]("\u00DF"), is(1));
assertThat([Link]("\u6771"), is(2));
assertThat([Link]("\uD801\uDC00"), is(3));

assertThat([Link](0), is('\u0041'));
assertThat([Link](1), is('\u00DF'));
assertThat([Link](2), is('\u6771'));
assertThat([Link](3), is('\uD801'));
assertThat([Link](4), is('\uDC00'));

assertThat([Link](0), is(0x0041));
assertThat([Link](1), is(0x00DF));
assertThat([Link](2), is(0x6771));
assertThat([Link](3), is(0x10400));
}

@Test
public void text() {

Text t = new Text("\u0041\u00DF\u6771\uD801\uDC00");

92 | Chapter 4: Hadoop I/O

 assertThat([Link](), is(10));

assertThat([Link]("\u0041"), is(0));
assertThat([Link]("\u00DF"), is(1));
assertThat([Link]("\u6771"), is(3));
assertThat([Link]("\uD801\uDC00"), is(6));

assertThat([Link](0), is(0x0041));
assertThat([Link](1), is(0x00DF));
assertThat([Link](3), is(0x6771));
assertThat([Link](6), is(0x10400));
}
}

String char

Text
indexOf() String char
find() Text
charAt() String char
code
PointAt() char
int charAt() Text
codePointAt() String

Iteration. Text

Text [Link]
bytesToCodePoint() Text
int
bytesToCodePoint()

public class TextIterator {

public static void main(String[] args) {

Text t = new Text("\u0041\u00DF\u6771\uD801\uDC00");

ByteBuffer buf = [Link]([Link](), 0, [Link]());

int cp;
while ([Link]() && (cp = [Link](buf)) != -1) {
[Link]([Link](cp));
}
}
}

Serialization | 93
 % hadoop TextIterator
41
df
6771
10400

Mutability. String Text Writable

NullWritable
Text set()
Text t = new Text("hadoop");
[Link]("pig");
assertThat([Link](), is(3));
assertThat([Link]().length, is(3));

getBytes()
getLength()
Text t = new Text("hadoop");
[Link](new Text("pig"));
assertThat([Link](), is(3));
assertThat("Byte length not shortened", [Link]().length, is(6));

getLength()
getBytes()

Resorting to String. Text

[Link] Text String
toString()
assertThat(new Text("hadoop").toString(), is("hadoop"));

BytesWritable
BytesWritable

00000002 03 05
BytesWritable b = new BytesWritable(new byte[] { 3, 5 });
byte[] bytes = serialize(b);
assertThat([Link](bytes), is("000000020305"));

BytesWritable set()
Text getBytes() Byte
sWritable
BytesWritable BytesWritable get
Length()
[Link](11);
assertThat([Link](), is(2));
assertThat([Link]().length, is(11));

94 | Chapter 4: Hadoop I/O

NullWritable
NullWritable Writable

NullWritable
NullWritable
SequenceFile

[Link]()

ObjectWritable and GenericWritable

ObjectWritable String
enum Writable null

ObjectWritable
SequenceFile
ObjectWritable ObjectWritable

GenericWritable

Writable collections
Writable [Link] Array
Writable TwoDArrayWritable MapWritable SortedMapWritable
ArrayWritable TwoDArrayWritable Writable
Writable
ArrayWritable TwoDArrayWritable

ArrayWritable writable = new ArrayWritable([Link]);

Writable SequenceFile
ArrayWritable
TwoDArrayWritable
public class TextArrayWritable extends ArrayWritable {
public TextArrayWritable() {
super([Link]);
}
}

ArrayWritable TwoDArrayWritable get() set()

toArray()

Serialization | 95
MapWritable SortedMapWritable [Link]<Writable,
Writable> [Link]<WritableComparable, Writable>

[Link]
Writable
MapWritable SortedMapWritable
byte
Writable MapWritable SortedMapWritable
MapWritable

MapWritable src = new MapWritable();

[Link](new IntWritable(1), new Text("cat"));
[Link](new VIntWritable(2), new LongWritable(163));

MapWritable dest = new MapWritable();

[Link](dest, src);
assertThat((Text) [Link](new IntWritable(1)), is(new Text("cat")));
assertThat((LongWritable) [Link](new VIntWritable(2)), is(new LongWritable(163)));

Writable
MapWritable SortedMapWritable
NullWritable Writable ArrayWritable
Writable
GenericWritable ArrayWritable
ListWritable MapWritable

Implementing a Custom Writable

Writable

Writable
Writable
Writable

Writable

Writable
TextPair

import [Link].*;

import [Link].*;

96 | Chapter 4: Hadoop I/O

public class TextPair implements WritableComparable<TextPair> {

private Text first;

private Text second;

public TextPair() {
set(new Text(), new Text());
}

public TextPair(String first, String second) {

set(new Text(first), new Text(second));
}

public TextPair(Text first, Text second) {

set(first, second);
}

public void set(Text first, Text second) {

[Link] = first;
[Link] = second;
}

public Text getFirst() {

return first;
}

public Text getSecond() {

return second;
}

@Override
public void write(DataOutput out) throws IOException {
[Link](out);
[Link](out);
}

@Override
public void readFields(DataInput in) throws IOException {
[Link](in);
[Link](in);
}

@Override
public int hashCode() {
return [Link]() * 163 + [Link]();
}

@Override
public boolean equals(Object o) {
if (o instanceof TextPair) {
TextPair tp = (TextPair) o;
return [Link]([Link]) && [Link]([Link]);
}
return false;
}

Serialization | 97
 @Override
public String toString() {
return first + "\t" + second;
}

@Override
public int compareTo(TextPair tp) {
int cmp = [Link]([Link]);
if (cmp != 0) {
return cmp;
}
return [Link]([Link]);
}
}

Text
first second
Writable
readFields()

write() readFields()
TextPair write() Text
Text readFields()
Text DataOutput
DataInput

Writable

hashCode() equals() toString() [Link] hash

Code() HashPartitioner

Writable TextOutputFormat
toString() TextOutputFormat
toString() Text
Pair Text

TextPair WritableComparable
compareTo()
TextPair TextArrayWrita
ble Text
TextArrayWritable Writable WritableComparable

98 | Chapter 4: Hadoop I/O

Implementing a RawComparator for speed
TextPair

TextPair
compareTo()
TextPair

TextPair Text
Text

Text Text RawComparator

TextPair

public static class Comparator extends WritableComparator {

private static final [Link] TEXT_COMPARATOR = new [Link]();

public Comparator() {
super([Link]);
}

@Override
public int compare(byte[] b1, int s1, int l1,
byte[] b2, int s2, int l2) {

try {
int firstL1 = [Link](b1[s1]) + readVInt(b1, s1);
int firstL2 = [Link](b2[s2]) + readVInt(b2, s2);
int cmp = TEXT_COMPARATOR.compare(b1, s1, firstL1, b2, s2, firstL2);
if (cmp != 0) {
return cmp;
}
return TEXT_COMPARATOR.compare(b1, s1 + firstL1, l1 - firstL1,
b2, s2 + firstL2, l2 - firstL2);
} catch (IOException e) {
throw new IllegalArgumentException(e);
}
}
}

static {
[Link]([Link], new Comparator());
}

WritableComparator RawComparator

Serialization | 99
 firstL1 firstL2
Text
decodeVIntSize() WritableUtils
readVInt()

TextPair

Custom comparators
TextPair

Writable [Link]
WritableUtils
RawComparator

TextPair First
Comparator
compare() compare()

public static class FirstComparator extends WritableComparator {

private static final [Link] TEXT_COMPARATOR = new [Link]();

public FirstComparator() {
super([Link]);
}

@Override
public int compare(byte[] b1, int s1, int l1,
byte[] b2, int s2, int l2) {

try {
int firstL1 = [Link](b1[s1]) + readVInt(b1, s1);
int firstL2 = [Link](b2[s2]) + readVInt(b2, s2);
return TEXT_COMPARATOR.compare(b1, s1, firstL1, b2, s2, firstL2);
} catch (IOException e) {
throw new IllegalArgumentException(e);
}
}

@Override
public int compare(WritableComparable a, WritableComparable b) {
if (a instanceof TextPair && b instanceof TextPair) {
return ((TextPair) a).[Link](((TextPair) b).first);

100 | Chapter 4: Hadoop I/O

 }
return [Link](a, b);
}
}

Serialization Frameworks
Writable

Serialization
[Link] WritableSerialization
Serialization Writable
Serialization Serializer
Deserializer

[Link]
Serialization [Link]
[Link] Writable

JavaSerialization

Integer String

Why Not Use Java Object Serialization?

Serialization | 101
 [Link]
[Link]

Serialization IDL

[Link]

102 | Chapter 4: Hadoop I/O

File-Based Data Structures

SequenceFile
SequenceFile

LongWrit
able Writable
SequenceFile
SequenceFile

SequenceFile

Writing a SequenceFile
SequenceFile createWriter()
[Link]
FSDataOutputStream FileSys
tem Path Configuration
Progressable
Metadata SequenceFile

SequenceFile Writable
Serialization

Serialization
Serialization

SequenceFile

File-Based Data Structures | 103

 [Link]
append() close() Sequence
[Link] [Link]
Sequence
File

public class SequenceFileWriteDemo {

private static final String[] DATA = {

"One, two, buckle my shoe",
"Three, four, shut the door",
"Five, six, pick up sticks",
"Seven, eight, lay them straight",
"Nine, ten, a big fat hen"
};

public static void main(String[] args) throws IOException {

String uri = args[0];
Configuration conf = new Configuration();
FileSystem fs = [Link]([Link](uri), conf);
Path path = new Path(uri);

IntWritable key = new IntWritable();

Text value = new Text();
[Link] writer = null;
try {
writer = [Link](fs, conf, path,
[Link](), [Link]());

for (int i = 0; i < 100; i++) {

[Link](100 - i);
[Link](DATA[i % [Link]]);
[Link]("[%s]\t%s\t%s\n", [Link](), key, value);
[Link](key, value);
}
} finally {
[Link](writer);
}
}
}

IntWritable Text
[Link] getLength()

% hadoop SequenceFileWriteDemo [Link]

[128] 100 One, two, buckle my shoe
[173] 99 Three, four, shut the door

104 | Chapter 4: Hadoop I/O

 [220] 98 Five, six, pick up sticks
[264] 97 Seven, eight, lay them straight
[314] 96 Nine, ten, a big fat hen
[359] 95 One, two, buckle my shoe
[404] 94 Three, four, shut the door
[451] 93 Five, six, pick up sticks
[495] 92 Seven, eight, lay them straight
[545] 91 Nine, ten, a big fat hen
...
[1976] 60 One, two, buckle my shoe
[2021] 59 Three, four, shut the door
[2088] 58 Five, six, pick up sticks
[2132] 57 Seven, eight, lay them straight
[2182] 56 Nine, ten, a big fat hen
...
[4557] 5 One, two, buckle my shoe
[4602] 4 Three, four, shut the door
[4649] 3 Five, six, pick up sticks
[4693] 2 Seven, eight, lay them straight
[4743] 1 Nine, ten, a big fat hen

Reading a SequenceFile

[Link]
next()
Writable next()

public boolean next(Writable key, Writable val)

true false

Writable

public Object next(Object key) throws IOException

public Object getCurrentValue(Object val) throws IOException

[Link]
next() null
getCurrentValue()
next() null

Writable Sequence
[Link] getKeyClass() getValueClass() ReflectionUtils

Writable

File-Based Data Structures | 105

public class SequenceFileReadDemo {

public static void main(String[] args) throws IOException {

String uri = args[0];
Configuration conf = new Configuration();
FileSystem fs = [Link]([Link](uri), conf);
Path path = new Path(uri);

[Link] reader = null;

try {
reader = new [Link](fs, path, conf);
Writable key = (Writable)
[Link]([Link](), conf);
Writable value = (Writable)
[Link]([Link](), conf);
long position = [Link]();
while ([Link](key, value)) {
String syncSeen = [Link]() ? "*" : "";
[Link]("[%s%s]\t%s\t%s\n", position, syncSeen, key, value);
position = [Link](); // beginning of next record
}
} finally {
[Link](reader);
}
}
}

[Link]

% hadoop SequenceFileReadDemo [Link]

[128] 100 One, two, buckle my shoe
[173] 99 Three, four, shut the door
[220] 98 Five, six, pick up sticks
[264] 97 Seven, eight, lay them straight
[314] 96 Nine, ten, a big fat hen
[359] 95 One, two, buckle my shoe
[404] 94 Three, four, shut the door
[451] 93 Five, six, pick up sticks
[495] 92 Seven, eight, lay them straight
[545] 91 Nine, ten, a big fat hen
[590] 90 One, two, buckle my shoe
...

106 | Chapter 4: Hadoop I/O

 [1976] 60 One, two, buckle my shoe
[2021*] 59 Three, four, shut the door
[2088] 58 Five, six, pick up sticks
[2132] 57 Seven, eight, lay them straight
[2182] 56 Nine, ten, a big fat hen
...
[4557] 5 One, two, buckle my shoe
[4602] 4 Three, four, shut the door
[4649] 3 Five, six, pick up sticks
[4693] 2 Seven, eight, lay them straight
[4743] 1 Nine, ten, a big fat hen

seek()

[Link](359);
assertThat([Link](key, value), is(true));
assertThat(((IntWritable) key).get(), is(95));

next()
[Link](360);
[Link](key, value); // fails with IOException

sync(long
position) [Link]
position
sync()

[Link](360);
assertThat([Link](), is(2021L));
assertThat([Link](key, value), is(true));
assertThat(((IntWritable) key).get(), is(59));

[Link] sync()

sync()
Syncable

Displaying a SequenceFile with the command-line interface

hadoop fs -text

File-Based Data Structures | 107

 toString()

% hadoop fs -text [Link] | head

100 One, two, buckle my shoe
99 Three, four, shut the door
98 Five, six, pick up sticks
97 Seven, eight, lay them straight
96 Nine, ten, a big fat hen
95 One, two, buckle my shoe
94 Three, four, shut the door
93 Five, six, pick up sticks
92 Seven, eight, lay them straight
91 Nine, ten, a big fat hen

Sorting and merging SequenceFiles

% hadoop jar $HADOOP_INSTALL/hadoop-*-[Link] sort -r 1 \

-inFormat [Link] \
-outFormat [Link] \
-outKey [Link] \
-outValue [Link] \
[Link] sorted
% hadoop fs -text sorted/part-00000 | head
1 Nine, ten, a big fat hen
2 Seven, eight, lay them straight
3 Five, six, pick up sticks
4 Three, four, shut the door
5 One, two, buckle my shoe
6 Nine, ten, a big fat hen
7 Seven, eight, lay them straight
8 Five, six, pick up sticks
9 Three, four, shut the door
10 One, two, buckle my shoe

108 | Chapter 4: Hadoop I/O

 [Link]
sort() merge()

The SequenceFile Format

SEQ

SequenceFile

File-Based Data Structures | 109

 writeInt() [Link]
Output Serialization

[Link]
[Link]

MapFile
MapFile SequenceFile MapFile
[Link]
Map

Writing a MapFile
MapFile SequenceFile
[Link] append()
IOException
WritableComparable Writable SequenceFile

110 | Chapter 4: Hadoop I/O

 MapFile
SequenceFile

public class MapFileWriteDemo {

private static final String[] DATA = {

"One, two, buckle my shoe",
"Three, four, shut the door",
"Five, six, pick up sticks",
"Seven, eight, lay them straight",
"Nine, ten, a big fat hen"
};

public static void main(String[] args) throws IOException {

String uri = args[0];
Configuration conf = new Configuration();
FileSystem fs = [Link]([Link](uri), conf);

IntWritable key = new IntWritable();

Text value = new Text();
[Link] writer = null;
try {
writer = new [Link](conf, fs, uri,
[Link](), [Link]());

for (int i = 0; i < 1024; i++) {

[Link](i + 1);
[Link](DATA[i % [Link]]);
[Link](key, value);
}
} finally {
[Link](writer);
}
}
}

MapFile
% hadoop MapFileWriteDemo [Link]

MapFile

% ls -l [Link]
total 104
-rw-r--r-- 1 tom tom 47898 Jul 29 22:06 data
-rw-r--r-- 1 tom tom 251 Jul 29 22:06 index

SequenceFile
% hadoop fs -text [Link]/data | head
1 One, two, buckle my shoe
2 Three, four, shut the door
3 Five, six, pick up sticks

File-Based Data Structures | 111

 4 Seven, eight, lay them straight
5 Nine, ten, a big fat hen
6 One, two, buckle my shoe
7 Three, four, shut the door
8 Five, six, pick up sticks
9 Seven, eight, lay them straight
10 Nine, ten, a big fat hen

% hadoop fs -text [Link]/index

1 128
129 6079
257 12054
385 18030
513 24002
641 29976
769 35947
897 41922

[Link]
setIndexInterval() [Link]

MapFile

MapFile

Reading a MapFile
MapFile
SequenceFile [Link] next()
false
public boolean next(WritableComparable key, Writable val) throws IOException

get()
public Writable get(WritableComparable key, Writable val) throws IOException

MapFile null
key key
val

MapFile

112 | Chapter 4: Hadoop I/O

 Text value = new Text();
[Link](new IntWritable(496), value);
assertThat([Link](), is("One, two, buckle my shoe"));

[Link]

getClosest() get()
null MapFile

MapFile boolean

MapFile

MapFile [Link]
0 1
2

Converting a SequenceFile to a MapFile

MapFile SequenceFile
SequenceFile MapFile
SequenceFile
SequenceFile
fix() MapFile
MapFile

public class MapFileFixer {

public static void main(String[] args) throws Exception {

String mapUri = args[0];

Configuration conf = new Configuration();

FileSystem fs = [Link]([Link](mapUri), conf);

Path map = new Path(mapUri);

File-Based Data Structures | 113

 Path mapData = new Path(map, MapFile.DATA_FILE_NAME);

// Get key and value types from data sequence file

[Link] reader = new [Link](fs, mapData, conf);
Class keyClass = [Link]();
Class valueClass = [Link]();
[Link]();

// Create the map file index file

long entries = [Link](fs, map, keyClass, valueClass, false, conf);
[Link]("Created MapFile %s with %d entries\n", map, entries);
}
}

fix()

MapFile

% hadoop jar $HADOOP_INSTALL/hadoop-*-[Link] sort -r 1 \

-inFormat [Link] \
-outFormat [Link] \
-outKey [Link] \
-outValue [Link] \
[Link] [Link]

% hadoop fs -mv [Link]/part-00000 [Link]/data

% hadoop MapFileFixer [Link]

Created MapFile [Link] with 100 entries

MapFile

114 | Chapter 4: Hadoop I/O