Hbase PPT PDF

Tutorial: HBase
Theory and Practice of a Distributed Data Store
Pietro Michiardi
Eurecom
Pietro Michiardi (Eurecom) Tutorial: HBase 1 / 102

Introduction
Introduction

Introduction RDBMS
Why yet another storage architecture?
Relational Databse Management Systems (RDBMS):

I Around since 1970s
I Countless examples in which they actually do make sense
The dawn of Big Data:

I Previously: ignore data sources because no cost-effective way to
store everything
F One option was to prune, by retaining only data for the last N days
I Today: store everything!
F Pruning fails in providing a base to build useful mathematical models

Introduction RDBMS
Batch processing
Hadoop and MapReduce:
I Excels at storing (semi- and/or un-) structured data
I Data interpretation takes place at analysis-time
I Flexibility in data classification
Batch processing: A complement to RDBMS:

I Scalable sink for data, processing launched when time is right
I Optimized for large file storage
I Optimized for “streaming” access
Random Access:
I Users need to “interact” with data, especially that “crunched” after a
MapReduce job
I This is historically where RDBMS excel: random access for
structured data

Introduction Column-Oriented DB
Column-Oriented Databases
Data layout:
I Save their data grouped by columns
I Subsequent column values are stored contiguously on disk
I This is substantially different from traditional RDBMS, which save
and store data by row
Specialized databases for specific workloads:

I Reduced I/O
I Better suited for compression → Efficient use of bandwidth
F Indeed, column values are often very similar and differ little
row-by-row
I Real-time access to data
Important NOTE:
I HBase is not a column-oriented DB in the typical term
I HBase uses an on-disk column storage format
I Provides key-based access to specific cell of data, or a sequential
range of cells
Introduction Column-Oriented DB
Column-Oriented and Row-Oriented storage layouts
Figure: Example of Storage Layouts

Introduction The problem with RDBMS
The Problem with RDBMS
RDBMS are still relevant

I Persistence layer for frontend application
I Store relational data
I Works well for a limited number of records
Example: Hush
I Used throughout this course
I URL shortener service
Let’s see the “scalability story” of such a service

I Assumption: service must run with a reasonable budget

Few thousands users: use a LAMP stack

I Normalize data
I Use foreign keys
I Use Indexes
Figure: The Hush Schema expressed as an ERD

TITLE
url
Find all short URLs for a given user

I JOIN user and shorturl tables
Stored Procedures
I Consistently update data from multiple clients
I Underlying DB system guarantees coherency
Transactions
I Make sure you can update tables in an atomic fashion
I RDBMS → Strong Consistency (ACID properties)
I Referential Integrity

Scaling up to tens of thousands of users

I Increasing pressure on the database server
I Adding more application servers is easy: they share their state on
the same central DB
I CPU and I/O start to be a problem on the DB
Master-Slave architecture
I Add DB server so that READS can be served in parallel
I Master DB takes all the writes (which are fewer in the Hush
application)
I Slaves DB replicate Master DB and serve all reads (but you need a
load balancer)

Scaling up to hundreds of thousands

I READS are still the bottlenecks
I Slave servers begin to fall short in serving clients requests
Caching
IAdd a caching layer, e.g. Memcached or Redis
IOffload READS to a fast in-memory system
→ You lose consistency guarantees
→ Cache invalidation is critical for having DB and Caching layer
consistent

Scaling up more
IWRITES are the bottleneck
IThe master DB is hit too hard by WRITE load
I Vertical scalability : beef up your master server
→ This becomes costly, as you may also have to replace your RDBMS
SQL JOINs becomes a bottleneck

I Schema de-normalization
I Cease using stored procedures, as they become slow and eat up a
lot of server CPU
I Materialized views (they speed up READS)
I Drop secondary indexes as they slow down WRITES

What if your application needs to further scale up?

I Vertical scalability vs. Horizontal scalability
Sharding
I Partition your data across multiple databases
FEssentially you break horizontally your tables and ship them to
different servers
F This is done using fixed boundaries
→ Re-sharding to achieve load-balancing

→ This is an operational nightmare
I Re-sharding takes a huge toll on I/O resources

Introduction NOSQL
Non-Relational DataBases
They originally do not support SQL

I In practice, this is becoming a thin line to make the distinction
I One difference is in the data model
I Another difference is in the consistency model (ACID and
transactions are generally sacrificed)
Consistency models and the CAP Theorem

I Strict: all changes to data are atomic
I Sequential: changes to data are seen in the same order as they
were applied
I Causal: causally related changes are seen in the same order
I Eventual: updates propagates through the system and replicas
when in steady state
I Weak: no guarantee

Introduction NOSQL
Dimensions to classify NoSQL DBs
Data model
I How the data is stored: key/value, semi-structured,
column-oriented, ...
I How to access data?
I Can the schema evolve over time?
Storage model
I In-memory or persistent?
I How does this affect your access pattern?
Consistency model
I Strict or eventual?
I This translates in how fast the system handles READS and WRITES
[2]

Introduction NOSQL
Physical Model
I Distributed or single machine?
I How does the system scale?
Read/Write performance
I Top-down approach: understands well the workload!
I Some systems are better for READS, other for WRITES
Secondary indexes
I Does your workload require them?
I Can your system emulate them?

Introduction NOSQL
Failure Handling
I How each data store handle server failures?
I Is it able to continue operating in case of failures?
F This is related to Consistency models and the CAP theorem
I Does the system support “hot-swap”?
Compression
I Is the compression method pluggable?
I What time of compression?
Load Balancing
I Can the storage system seamlessly balance load?

Introduction NOSQL
Atomic read-modify-write
I Easy in a centralized system, difficult in a distributed one
I Prevent race conditions in multi-threaded or shared-nothing designs
I Can reduce client-side complexity
Locking, waits and deadlocks

I Support for multiple client accessing data simultaneously
I Is locking available?
I Is it wait-free, hence deadlock free?
Impedance Match
“One-size-fits-all” has been long dismissed: need to find the perfect
match for your problem.

Introduction Denormalization
Database (De-)Normalization
Schema design at scale

I A good methodology is to apply the DDI principle [8]
F Denormalization
F Duplication
F Intelligent Key design
Denormalization
I Duplicate data in more than one table such that at READ time no
further aggregation is required
Next: an example based on Hush
I How to convert a classic relational data model to one that fits
HBase

Example: Hush - from RDBMS to HBase
Figure:TITLE
The Hush Schema expressed as an ERD
url
shorturl table: contains the short URL

shorturl url shorturl click
click table: contains click tracking, and other statistics,
refShortId
aggregated on a daily basis (essentially, a counter)
shortId refShortId
user table: contains user information
URL table: http://hush.li/a23eg
contains a replica of the page linked to a short URL,
a23eg
including META data and content (this is done for batch analysis
shorturl
purposes)
Figure:TITLE
url
user table is indexed

shorturl
on the username
url shorturl
field,
click
for fast user lookup
refShortId
shorturl table is indexed on the short URL (shortId) field, for
fast shortshortId
URL refShortId
lookup
http://hush.li/a23eg
a23eg
Pietro Michiardi (Eurecom) shorturl HBase

Tutorial: 21 / 102
Figure:TITLE
url
shorturl and user tables are related through a foreign key

relation on the shorturl
userId url shorturl click
URL table is related to shorturl table withrefShortId a foreign key on the

URL id shortId refShortId
click table http://hush.li/a23eg
is related to shorturl table with a foreign key on the
short URL id a23eg
NOTE: a web page is stored only once (even if multiple users link
shorturl
to it), but each users maintain separate statistics

shorturl table: stores each

short URL, usage statistics
(various time-ranges in
separate column-families with
distinct TTL settings)
I Note the dimensional postfix
appended to the time
information
url table: stores the

downloaded page, and the
extracted details
I This table uses compression
Figure: The Hush Schema in HBase

user-shorturl
user-shorturl table: this is

a lookup table (basically an
index) to find all shortIDs for a
given user
I Note that this table is filled
at insert time, it’s not
automatically generated by
HBase
user table: stores user details
Figure: The Hush Schema in HBase

user-shorturl
Example: Hush - RDBMS vs HBase
Same number of tables

I Their meaning is different
I click table has been absorbed by the shorturl table
I statistics are stored with the date as the key, so that they can be
accessed sequentially
I The user-shorturl table is replacing the foreign key
relationship, making user-related lookups faster
Normalized vs. De-normalized data

I Wide tables and column-oriented design eliminates JOINs
I Compound keys are essential
I Data partitioning is based on keys, so a proper understanding
thereof is essential

Introduction HBase Sketch
HBase building blocks
The backdrop: BigTable

I GFS, The Google FileSystem [6]
I Google MapReduce [4]
I BigTable [3]
What is BigTable?
I BigTable is a distributed storage system for managing structured
data designed to scale to a very large size
I BigTable is a sparse, distributed, persistent multi-dimensional
sorted map
What is HBase?
I Essentially it’s an open-source version of BigTable
I Differences listed in [5]


Tables, Rows, Columns, and Cells
The most basic unit in HBase is a column
I Each column may have multiple versions, with each distinct value
contained in a separate cell
I One or more columns form a row, that is addressed uniquely by a
row key
A table is a collection of rows

I All rows are always sorted lexicographically by their row key
hbase(main):001:0> scan 'table1'

ROW COLUMN+CELL
row-1 column=cf1:, timestamp=1297073325971 ...
row-abc column=cf1:, timestamp=1297073349875 ...
7 row(s) in 0.1100 seconds

Lexicographical ordering of row keys

I Keys are compared on a binary level, byte by byte, from left to right
I This can be thought of as a primary index on the row key!
I Row keys are always unique
I Row keys can be any arbitrary array of bytes
Columns
I Rows are composed of columns
I Can have millions of columns
I Can be compressed or tagged to stay in memory


Column Families
IColumns are grouped into column families
→ Semantical boundaries between data
I Column families and columns stored together in the same low-level
storage file, called an HFile

I Defined when table is created
I Should not be changed too often

I The number of column families should be reasonable [WHY?]
I Column family name composed by printable characters
References to columns
I Column “name” is called qualifier, and can be any arbitrary
number of bytes
I Reference: family:qualifier

A note on the NULL value

I In RDBMS NULL cells need to be set and occupy space
I In HBase, NULL cells or columns are simply not stored
A cell
I Every column value, or cell, is timestamped (implicitly or explicitly)
F This can be used to save multiple versions of a value that changes
over time
F Versions are stored in decreasing timestamp, most recent first
I Cell versions can be constrained by predicate deletions
F Keep only values from the last week


Access to data
I (Table, RowKey, Family, Column, Timestamp) →
Value
I SortedMap<RowKey, List<SortedMap<Column,
List<Value, Timestamp»»
IRow data access is atomic and includes any number of columns

IThere is no further guarantee or transactional feature spanning
multiple rows
→ HBase is strictly consistent
Which means:
I The first SortedMap is the table, containing a List of column
families
I The families contain another SortedMap, representing columns
and a List of value, timestamp tuples
Automatic Sharding
Region
I This is the basic unit of scalability and load balancing
I Regions are contiguous ranges of rows stored together → they are
the equivalent of range partitions in sharded RDBMS
I Regions are dynamically split by the system when they become too
large
I Regions can also be merged to reduce the number of storage files
Regions in practice
I Initially, there is one region
I System monitors region size: if a threshold is attained, SPLIT
F Regions are split in two at the middle key
F This creates roughly two equivalent (in size) regions

Automatic Sharding
Region Servers
I Each region is served by exactly one Region Server
I Region servers can serve multiple regions
I The number of region servers and their sizes depend on the
capability of a single region server
Server failures
I Regions allow for fast recovery upon failure
I Fine-grained Load Balancing is also achieved using regions as they
can be easily moved across servers

Storage API
No support for SQL

I CRUD operations using a standard API, available for many “clients”
I Data access is not declarative but imperative
Scan API
I Allows for fast iteration over ranges of rows
I Allows to limit the number and which column are returned
I Allows to control the version number of each cell
Read-modify-write API
I HBase supports single-row transactions
I Atomic read-modify-write on data stored in a single row key

Storage API
Counters
IValues can be interpreted as counters and updated atomically
ICan be read and modified in one operation
→ Implement global, strictly consistent, sequential counters
Coprocessors
I These are equivalent to stored-procedures in RDBMS
I Allow to push user code in the address space of the server
I Access to server local data
I Implement lightweight batch jobs, data pre-processing, data
summarization


HBase implementation
Data Storage
I Store files are called HFiles
I Persistent and ordered immutable maps from key to value
I Internally implemented as sequences of blocks with an index at the
end
I Index is loaded when the HFile is opened and kept in memory
Data lookups
I Since HFiles have a block index, lookup can be done with a single
disk seek
I First, the block possibly containing a given lookup key is determined
with a binary search in the in-memory index
I Then a block read is performed to find the actual key
Underlying file system

I Many are supported, usually HBase deployed on top of HDFS

WRITE operation
I First, data is written to a commit log, called WAL (write-ahead-log)
I Then data is moved into memory, in a structure called memstore
I When the size of the memstore exceeds a given threshold it is
flushed to an HFile to disk
How can HBase write, while serving READS and WRITES?

I Rolling mechanism
F new/empty slots in the memstore take the updates
F old/full slots
I Note that data in memstore is sorted by keys, matching what
happens in the HFiles
Data Locality
I Achieved by the system looking up for server hostnames
I Achieved through intelligent key design
Deleting data
I Since HFiles are immutable, how can we delete data?
I A delete marker (also known as tombstone marker ) is written to
indicate that a given key is deleted
I During the read process, data marked as deleted is skipped
I Compactions (see next slides) finalize the deletion process
READ operation
I Merge of what is stored in the memstores (data that is not on disk)
and in the HFiles
I The WAL is never used in the READ operation
I Several API calls to read, scan data


Compactions
IFlushing data from memstores to disk implies the creation of new
HFiles each time
→ We end up with many (possibly small) files
→ We need to do housekeeping [WHY?]
Minor Compaction
I Rewrites small HFiles into fewer, larger HFiles
I This is done using an n-way merge1
Major Compaction
I Rewrites all files within a column family or a region in a new one
I Drop deleted data
I Perform predicated deletion (e.g. delete old data)
1
What is MergeSort?
HBase: a glance at the architecture
Master node: HMaster

I Assigns regions to region servers using ZooKeeper
I Handles load balancing
I Not part of the data path
I Holds metadata and schema
Region Servers
I Handle READs and WRITEs
I Handle region splitting
Architecture
Architecture

Architecture Seek vs. Transfer
Seek vs. Transfer
Fundamental difference between RDBMS and alternatives

I B+Trees
I Log-Structured Merge Trees
Seek vs. Transfer

I Random access to individual cells
I Sequential access to data

B+ Trees
Dynamic, multi-level indexes

I Efficient insertion, lookup and deletion
I Q: What’s the difference between a B+ Tree and a Hash Table?
I Frequent updates may imbalance the trees → Tree optimization
and re-organization is required (which is a costly operation)
Bounds on page size

I Number of keys in each branch
I Larger fanout compared to binary trees
I Lower number of I/O operations to find a specific key
Support for range scans

I Leafs are linked and represent an in-order list of all keys
I No costly tree-traversal algorithms required

LSM-Trees
Data flow
I Incoming data is first stored in a logfile, sequentially
I Once the log has the modification saved, data is pushed in memory
F In-memory store holds most recent updates for fast lookup
I When memory is “full”, data is flushed in a store file to disk, as a
sorted list of key → record pair
I At this point, the log file can be thrown away
How store files are arranged

I Similar idea of a B+ Tree, but optimized for sequential disk access
I All nodes of the tree try to be filled up completely
I Updates are done in a rolling merge fashion
F The system packs existing on-disk multi-page blocks with in-memory
data until the block reaches full capacity

LSM-Trees
Clean-up process
I As flushes take place over time, a lot of store files are created
I Background process aggregates files into larger ones to limit disk
seeks
I All store files are always sorted by key → no re-ordering required to
fit new keys in
Data Lookup
I Lookups are done in a merging fashion
F First lookup in the in-memory store
F If miss, the lookup in the on-disk store
Deleting data
I Use a delete marker
I When pages are re-written, deleted markers and keys are
eventually dropped
I Predicate deletion happens here
B+ Tree vs. LSM-Trees
B+ Tree [1]
I Work well when there are not so many updates
I The more and the faster you insert data at random locations the
faster pages get fragmented
I Updates and deletes are done at disk seek rates, rather than
transfer rates
LSM-Tree [7]
I Work at disk transfer rate and scale better to huge amounts of data
I Guarantee a consistent insert rate
F They transform random into sequential writes
I Reads are independent from writes
I Optimized data layout which offers predictable boundaries on disk
seeks

Architecture Storage
Storage
Overview
Figure: Overview of how HBase handles files in the filesystem
.META.
Storage
Overview
HBase handles two kinds of file types
I One is used for the WAL
I One is used for the actual data storage
Who does what

I HMaster
F Low-level operations
F Assign region servers to key space
F Keep metadata
F Talk to ZooKeeper
I HRegionServer
F Handles the WAL and HFiles
F These files are divided in to blocks and stored into HDFS
F Block size is a parameter

Storage
Overview
General communication flow
I A client contacts ZooKeeper when trying to access a particular row
I Recovers from ZooKeeper the server name that host the -ROOT-
region
I Using the -ROOT- information the client retrieves the server name
that host the .META. table region
F The .META. table region contain the row key in question
I Contact the reported .META. server and retrieve the server name
that has the region containing the row key in question
Caching
I Generally, lookup procedures involve caching row key locations for
faster subsequent lookups

Storage
Overview
Important Java Classes
I HRegionServer handles one or more regions and create the
corresponding HRegion object
I When an HRegion object is opened it creates aStore instance for
each HColumnFamily
I Each Store instance can have:
F One or more StoreFile instances
F A MemStore instance
I HRegionServer has a shared HLog instance

Storage
Write Path
External client insert data in HBase
I Issues an HTable.put(Put) request to HRegionServer
I HRegionServer hands the request to the HRegion instance that
matches the request [Q: What is the matching criteria?]
How the system reacts to a write request

I Write data to the WAL, represented by the HLog class
F The WAL stores HLogKey instances in a HDFS SequenceFile
F These keys contain a sequence number and the actual data
F In case of failure, this data can be used to replay not-yet-persisted
data
I Copy data in theMemStore
F Check if MemStore size has reached a threshold
F If yes, launch a flush request
F Launch a thread in the HRegionServer and flush MemStore data to
an HFile
Storage
HBase Files
What and where are HBase files (including WAL, HFile,...)
stored?
I HBase has a root directory set to “/hbase” in HDFS
I Files can be divided into:
F Those that reside under the HBase root directory
F Those that are in the per-table directories
/hbase
I .logs
I .oldlogs
I .hbase.id
I .hbase.version
I /example-table

Storage
HBase Files
/example-table
I .tableinfo
I .tmp
I “...Key1...”
F .oldlogs
F .regioninfo
F .tmp
F colfam1/
colfam1/
I “....column-key1...”

Storage
HBase: Root-level files

.logs directory
I WAL files handled by HLog instances
I Contains a subdir for each HRegionServer
I Each subdir contains many HLog files
I All regions from that HRegionServer share the same HLog files
.oldlogs directory
I When data is persisted to disk (from Memstores) log files are
decommissioned to the .oldlogs dir
hbase.id and hbase.version

I Represent the unique ID of the cluster and the file format version

Storage
HBase: Table-level files

Every table has its own directory
I .tableinfo: stores the serialized HTableDescriptor
F This include the table and column family schema
I .tmp directory
F Contains temporary data

Storage
HBase: Region-level files
Inside each table dir, there is a separate dir for every region
in the table
The name of each of this dirs is the MD5 hash of a region name
I Inside each region there is a directory for each column family
I Each column family directory holds the actual data files, namely
HFiles
I Their name is just an arbitrary random number
Each region directory also has a .regioninfo file
I Contains the serialized information of the HRegionInfo instance
Split Files
I Once the region needs to be split, a splits directory is created
F This is used to stage two daughter regions
F If split is successful, daughter regions are moved up to the table
directory
Storage
HBase: A note on region splits
Splits triggered by store file (region) size

I Region is split in two
I Region is closed to new requests
I .META. is updated
Daughter regions initially reside on the same server

I Both daughters are compacted
I Parent is cleaned up
I .META. is updated
Master schedules new regions to be moved off to other

servers

hbase.hstore.compaction.min
hbase.hstore.compactionThreshold
Storage 3 2
HBase: Compaction
10
Process that takes care of re-organizing store files
hbase.hstore.compaction.max
I Essentially to conform to underlying filesystem requirements
hbase.hstore.compaction.min.size
I Compaction check hbase.hstore.compaction.max.size
when memstore is flushed Long.MAX_VALUE
Minor and Major compactions

I Always from the oldest to the newest files
I Avoid all servers to perform compaction concurrently
Figure: A set of store files showing the minimum compaction threshold
hbase.hstore.compaction.ratio 1.2
Storage
HFile format
Store files are implemented by the HFile class
I Efficient data storage is the goal
HFiles consist of a variable number of blocks

I Two fixed blocks: info and trailer
I index block: records the offsets of the data and meta blocks
I Block size: large → sequential access; small → random access
Figure: The HFile structure

Storage
HFile size and HDFS block size
HBase uses any underlying filesystem
In case HDFS is used

IHDFS block size is generally 64MB
IThis is 1,024 times the default HFile block size (64 KB)
→ There is no correlation between HDFS block and HFile sizes

Storage
The KeyValue Format
Each KeyValue in the HFile is a low-level byte array
I It allows for zero-copy access to the data
Format
I Fixed-length preambule indicated the length of the key and value
F This is useful to offset into the array to get direct access to the value,
ignoring the key
I Key format
F Contains row key, column family name, column qualifier...
F [TIP]: consider small keys to avoid overhead when storing small
data
Figure: The KeyValue Format

Architecture WAL
The Write-Ahead Log
Main tool to ensure resiliency to failures

I Region servers keep data in-memory until enough is collected to
warrant a flush
I What if the server crashes or power is lost?
WAL is a common approach to address fault-tolerance

I Every data update is first written to a log
I Log is persisted (and replicated, since it resides on HDFS)
I Only when log is written, client is notified a successful operation on
data

Architecture WAL
The Write-Ahead Log
Figure: The write path of HBase
put() delete() increment()

WAL recordsKeyValue
all changes to data
HRegionServer
I Can be replayed in case of server failure
KeyValue HRegion
I If write to WAL fails, the whole operations has to fail
MemStore Store
Architecture WAL
The Write-Ahead Log
Write Path
I Client modifies data (put(), delete(), increment())
put() delete() increment()
I Modifications are wrapped into a KeyValue object
KeyValue
I Objects are batched HRegionServer
to the corresponding HRegionServer
I Objects are
KeyValue routed to the corresponding HRegion
HRegion
I Objects are written to WAL and in the MemStore
MemStore Store
Architecture Read Path
Read Path
HBase uses multiple store files per column family

I These can be either in-memory and/or materialized on disk
I Compactions and clean-up background processes take care of
store files maintenance
I Store files are immutable, so deletion is handled in a special way
The anatomy of a get command

I HBase uses a QueryMatcher in combination with a
ColumnTracker
I First, an exclusion check is performed to filter skip files (and
eventually tombstone labelled data)
I Scanning data is implemented by a RegionScanner class which
retrieves a StoreScanner
I StoreScanner includes both the MemStore and HFiles
I Read/Scans happen in the same order as data is saved

Architecture Region Lookups
Region Lookups
How does a client find the region server hosting a specific

row key range?
I HBase uses two special catalog tables, -ROOT- and .META.
I The -ROOT- table is used to refer to all regions in the .META. table
Three-level B+ Tree -like operation

I Level 1: a node stored in ZooKeeper, containing the location
(region server) of the -ROOT- table
I Level 2: Lookup in the -ROOT- table to find a matching meta region
I Level 3: Retrieve the table region from the .META. table

Region Lookups
Where to send requests when looking for a specific row key?

I This information is cached, but the first time or when the cache is
stale or when there is a miss due to compaction, the following
procedure applies
Recursive discovery process

I Ask the region server hosting the matching .META. table to retrieve
the row key address
I If the information is invalid, it backs out: asks the -ROOT- table
where the relevant .META. region is
I If this fails, ask ZooKeeper where the -ROOT- table is

Region Lookups

Key Design
Key Design

Key Design Concepts
Concepts
HBase has two fundamental key structures

I Row key
I Column key
Both can be used to convey meaning

I Because they store particularly meaningful data
I Because their sorting order is important

Key Design Concepts
Concepts
Logical vs. on-disk layout of a table
I Main unit of separation within a table is the column family
I The actual columns (as opposed to other column-oriented DB) are
not used to separate data
I Although cells are stored logically in a table format, rows are stored
as linear sets of the cells
I Cells contain all the vital information inside them

Key Design Concepts
Concepts
Logical Layout (Top-Left)

ITable consists of rows and columns
IColumns are the combination of a column family name and a
column qualifier
→ <cf name: qualifier> is the column key
I Rows have a row key to address all columns of a single logical row
KeyValue
Pietro Michiardi (Eurecom) Tutorial: HBase KeyValue 72 / 102

Key Design Concepts
Concepts
Folding the Logical Layout (Top-Right)

IThe cells of each row are stored one after the other
IEach column family are stored separately
→ On disk all cells of one family reside on an individual StoreFile
I HBase does not store unset cells
→ Row and column key is required to address every cell

KeyValue

Key Design Concepts
Concepts
Versioning
IMultiple versions of the same cell stored consecutively, together
with the timestamp
I Cells are sorted in descending order of timestamp
→ Newest value first
KeyValue object
I The entire cell, with all the structural information, is a KeyValue
object
I Contains: row key, <column family: qualifier> →
column key, timestamp and value
I Sorted by row key first, then by column key

Key Design Concepts
Concepts
Physical Layout (Lower-Right)

I Select data by row key
F This reduces the amount of data to scan for a row or a range of rows
I Select data by row key and column key
F This focuses the system on an individual storage file
I Select data by column qualifier
KeyValue
F Exact lookups, including filters to omit useless data
Key Design Concepts
Concepts
KeyValue
Summary of key lookup properties
KeyValue

Key Design Tall-Narrow vs. Flat-Wide
Tall-Narrow vs. Flat-Wide Tables
Tall-Narrow Tables
I Few columns
I Many rows
Flat-Wide Tables
I Many columns
I Few rows
Given the query granularity explained before

→ Store parts of the cell data in the row key
I Furthermore, HBase splits at row boundaries
→ It is recommended to go for Tall-Narrow Tables

Key Design Tall-Narrow vs. Flat-Wide
Tall-Narrow vs. Flat-Wide Tables

Example: email data - version 1
IYou have all emails of a user in a single row (e.g. userID is the
row key)
I There will be some outliers with orders of magnitude more emails
than others
→ A single row could outgrow the maximum file/region size and work
against split facility
Example: email data - version 2

I Each email of a user is stored in a separate row (e.g.
userID:messageID is the row key)
I On disk this makes no difference (see the disk layout figure)
F If the messageID is in the column qualifier or the row key, each cell
still contains a single email message
→ The table can be split easily and the query granularity is more
fine-grained

Key Design Partial Key Scans
Partial Key Scans
Partial Key Scans reinforce the concept of Tall-Narrow Tables

I From the email example: assume you have a separate row per
message, across all users
I If you don’t have an exact combination of user and message ID you
cannot access a particular message
Partial Key Scan solves the problems

ISpecify a start and end key
IThe start key is set to the exact userID only, with the end key set
at userID+1
→ This triggers the internal lexicographic comparison mechanism
F Since the table does not have an exact match, it positions the scan at:
<userID>:<lowest-messageID>
I The scan will then iterate over all the messages of an exact user,
parse the row key and get the messageID

Key Design Partial Key Scans
Partial Key Scans
Composite keys and atomicity

I Following the email example: a single user inbox now spans many
rows
I It is no longer possible to modify a single user inbox in one atomic
operation
If this is acceptable or not, depends on the application at

hand

Key Design Time Series Data
Time Series Data
Stream processing of events

IE.g. data coming from a sensor, stock exchange, monitoring
system ...
I Such data is a time series → The row key represents the event
time
→ HBase will store all rows sorted in a distinct range, namely regions
with specific start and stop keys
Sequential monotonously increasing nature of time series

data
IAll incoming data is written to the same region (and hence the
same server)
→ Regions become HOT!
I Performance of the whole cluster is bound to that of a single
machine

Time Series Data
Solution to achieve load balancing: Salting

I We want data to be spread over all region servers
I This can be done, e.g., by prefixing the row key with a
non-sequential number
Salting example
byte prefix = (byte) (Long.hashCode(timestamp) % <number of
region servers>);
byte[] rowkey = Bytes.add(Bytes.toBytes(prefix),
Bytes.toBytes(timestamp));
- Data access needs to be fanned out across many servers

+ Use multiple threads to read for I/O performance: e.g. use the
Map phase of MapReduce

Time Series Data
Solution to achieve load balancing: Field swap/promotion

I Move the timestamp filed of the row key or prefix it with another
field
F If you already have a composite row key, simply swap elements
F Otherwise if you only have the timestamp, you need to promote
another field
I The sequential, monotonously increasing timestamp is moved to a
secondary position in the row key
- You can only access data (especially time ranges) for a given
swapped or promoted field (but this could be a feature)
+ You achieve load balancing

Time Series Data
Solution to achieve load balancing: Randomization

I byte[] rowkey = MD5(timestamp)
I This gives you a random distribution of the row key across all
available region servers
- Less than ideal for range scans

+ Since you can re-hash the timestamp, this solution is good for
random access

Time Series Data
Summary

MapReduce Integration
MapReduce Integration

MapReduce Integration Recap
Introduction
In the following we review the main classes involved in

reading and writing data from/to an underlying data store
For MapReduce to work with HBase, some more practical

issues have to be addressed
I E.g.: creating an appropriate JAR file inclusive of all required
libraries
Refer to [5], Chapter 7 for an in-depth treatment of this subject

Main classes involved in MapReduce
Figure: Main MapReduce Classes

Classes

InputFormat
It is responsible for two things

I Splits input data
I Returns a RecordReader instance
F Defines a key and a value object
F Provides a next() method to iterate over input records
Figure: InputFormat hierarchy

TableInput
FormatBase TableInputFormat
InputFormat → TableInputFormatBase
Implement a full turnkey solution to scan an HBase table

I Splits the table into proper blocks and hand them to the
MapReduce process
Must supply a Scan instance to interact with a table

I Specify start and stop keys for the scan
I Add filters (optional)
I Specify the number of versions

TableInputFormat MapReduce Integration Recap
Main classesScan
involved in MapReduce
Mapper
TableInputFormat
Each record read using the RecordReader is processed

using
Mapper
the map() method
Mapper
RecordReader
The Mappermap()
reads specific types of input key/value
Mapper pairs, but
emit possibly another type
Figure: The Mapper hierarchy

TableMapper ImmutableBytes
Writable Result
TableRecordReader
Mapper → TableMapper
TableMapper class enforces:

I The input key to the mapper to be an ImmutableBytesWritable
type
I The input value to be a Result type
A handy implementation is the IdentityTableMapper

I This is the equivalent of an identity mapper

OutputFormat
OutputFormat
Used to persist data
I OutputFormat
Output written to files
I Output written to HBase tables
TableOutputFormat TableRecord
Writer
F This is done using a TableRecordWriter
Figure: The OutputFormat hierarchy

Mapper
Reducer OutputFormat
Reducer
OutputFormat → TableOutputFormat
This is the class that handles the key/valu pairs and writes
them to their final destination
I Single instance that takes the output record from each reducer
subsequently
Details
I Must specify the table name when the MR job is created
I Handles buffer flushing implicitly (autoflush option is set to false)

MapReduce Locality
How does the system make sure data is placed close to

where it is needed?
I This is done implicitly by MapReduce when using HDFS
I When MapReduce uses HBase things are a bit different
How HBase handles data locality

IShared vs. non-shared cluster
IHBase store its files on HDFS (HFiles and WAL)
I HBase servers are not restarted frequently and they perform
compactions regularly
→ HDFS is smart enough to ensure data locality
F There is a block placement policy that enforces local writes
F The data node compares the server name of the writer with its own
F If they match, the block is written to the local filesystem
I Just be careful about region movements during load balancing or
server failures

Table Splits
When running a MapReduce job that reads from an HBase

table you use the TableInputFormat
I Overrides getSplits() and createRecordReader()
Before a job is run, the framework calls getSplit() to

determine how the data is to be separated into chunks
ITableInputFormat, given the Scan instance you define, divide
the table at region boundaries
→ The number of input splits is equal to all regions between the start
and stop keys

Table Splits
When a job starts, the framework calls
createRecordReader() for each input split
I It iterates over the splits and create a new TableRecordReader
with the current split
I Each TableRecordReader handles exactly one region, reading
and mapping every row from the region’s start and end keys
Data locality
I Each split contains the server name hosting the region
I The framework checks the server name and if the TaskTracker is
running on the same machine, it will run it on that server
I The RegionServer is colocated with the HDFS DataNode, hence
data is read from the local filesystem
TIP: Turn off speculative execution!

References
References I
[1] B+ tree.
http://en.wikipedia.org/wiki/B%2B_tree.
[2] Eric Brewer.
Lessons from giant-scale services.
In In IEEE Internet Computing, 2001.
[3] Fay Chang, Jeffrey Dean, Sanjay Ghemawat, Wilson C. Hsieh,
Deborah A. Wallach, Mike Burrows, Tushar Chandra, Andrew
Fikes, and Robert E. Gruber.
Bigtable: A distributed storage system for structured data.
In Proc. od USENIX OSDI, 2006.
[4] Jeffrey Dean and Sanjay Ghemawat.
Mapreduce: Simplified data processing on large clusters.
In Proc. of ACM OSDI, 2004.

References
References II
[5] Lars George.

HBase, The Definitive Guide.
O’Reilly, 2011.
[6] Sanjay Ghemawat, Howard Gobioff, and Shun-Tak Leung.
The google file system.
In Proc. of ACM OSDI, 2003.
[7] Patrick O’Neil, Edward Cheng, Dieter Gawlick, and Elizabeth
O’Neil.
The log-structured merge-tree (lsm-tree).
1996.

References
References III
[8] D. Salmen.
Cloud data structure diagramming techniques and design patterns.
https://www.data-tactics-corp.com/index.php/
component/jdownloads/finish/22-white-papers/
68-cloud-data-structure-diagramming, 2009.

Hbase PPT PDF

Uploaded by

Document Informationclick to expand document information

Copyright:

Available Formats

Hbase PPT PDF

Uploaded by

Document Information

Original Title

Copyright

Available Formats

Share this document

Share or Embed Document

Sharing Options

Did you find this document useful?

Is this content inappropriate?

Copyright:

Available Formats

Hbase PPT PDF

Uploaded by

Copyright:

Available Formats

Tutorial: HBase

Theory and Practice of a Distributed Data Store

Pietro Michiardi (Eurecom) Tutorial: HBase 1 / 102

Pietro Michiardi (Eurecom) Tutorial: HBase 2 / 102

Why yet another storage architecture?

Relational Databse Management Systems (RDBMS):

The dawn of Big Data:

Pietro Michiardi (Eurecom) Tutorial: HBase 3 / 102

Batch processing: A complement to RDBMS:

Pietro Michiardi (Eurecom) Tutorial: HBase 4 / 102

Specialized databases for specific workloads:

Column-Oriented and Row-Oriented storage layouts

Figure: Example of Storage Layouts

Pietro Michiardi (Eurecom) Tutorial: HBase 6 / 102

The Problem with RDBMS

RDBMS are still relevant

Let’s see the “scalability story” of such a service

Pietro Michiardi (Eurecom) Tutorial: HBase 7 / 102

The Problem with RDBMS

Few thousands users: use a LAMP stack

Figure: The Hush Schema expressed as an ERD

The Problem with RDBMS

Find all short URLs for a given user

Pietro Michiardi (Eurecom) Tutorial: HBase 9 / 102

The Problem with RDBMS

Scaling up to tens of thousands of users

Pietro Michiardi (Eurecom) Tutorial: HBase 10 / 102

The Problem with RDBMS

Scaling up to hundreds of thousands

Pietro Michiardi (Eurecom) Tutorial: HBase 11 / 102

The Problem with RDBMS

SQL JOINs becomes a bottleneck

Pietro Michiardi (Eurecom) Tutorial: HBase 12 / 102

The Problem with RDBMS

What if your application needs to further scale up?

→ Re-sharding to achieve load-balancing

Pietro Michiardi (Eurecom) Tutorial: HBase 13 / 102

They originally do not support SQL

Consistency models and the CAP Theorem

Pietro Michiardi (Eurecom) Tutorial: HBase 14 / 102

Dimensions to classify NoSQL DBs

Pietro Michiardi (Eurecom) Tutorial: HBase 15 / 102

Dimensions to classify NoSQL DBs

Pietro Michiardi (Eurecom) Tutorial: HBase 16 / 102

Dimensions to classify NoSQL DBs

Pietro Michiardi (Eurecom) Tutorial: HBase 17 / 102

Dimensions to classify NoSQL DBs

Locking, waits and deadlocks

Pietro Michiardi (Eurecom) Tutorial: HBase 18 / 102

Schema design at scale

Pietro Michiardi (Eurecom) Tutorial: HBase 19 / 102

Example: Hush - from RDBMS to HBase

shorturl table: contains the short URL

Example: Hush - from RDBMS to HBase

user table is indexed

Pietro Michiardi (Eurecom) shorturl HBase

Example: Hush - from RDBMS to HBase

shorturl and user tables are related through a foreign key