Showing posts with label hadoop. Show all posts
Showing posts with label hadoop. Show all posts

Thursday, December 18, 2014

Useful links

Tuesday, December 20, 2011

Realtime Apache Hadoop at Facebook


  1. Facebook use Hadoop/Hbase for a following products Titan( Facebook messages), Puma( real-time publisher insights), ODS( Facebook internal metrics).

  2. Currently in use 4 data centers, about 2000 nodes.

  3. Data center replication is not supported as part of HBase, so they wrote some internal product to do that.

  4. All data LZO compressed.

  5. Each data center has 2PB of data replicated x3, total 6PB.

  6. Each node have 12 discs each 1Tb.

  7. Load on each node 55% read, 45% write.

  8. To collect data from front end servers they're using Scribe, which is open sourced.

Slides here.

WHY HADOOP AND HBASE( from facebook paper)


The requirements for the storage system for our workloads can be summarized as follows:



1. Elasticity: We need to be able to add incremental capacity to our storage systems with minimal overhead and no downtime. In some cases we may want to add capacity rapidly and the system should automatically balance load and utilization across new hardware.

2. High write throughput: Most of the applications store (and optionally index) tremendous amounts of data and require high aggregate write throughput.

3. Efficient and low-latency strong consistency semantics within a data center:
There are important applications like Messages that require strong consistency within a data center. This requirement often arises directly from user expectations. For example ‘‘unread’’ message counts displayed on the home page and the messages shown in the inbox page view should be consistent with respect to each other. While a globally distributed strongly consistent system is practically impossible, a system that could at least provide strong consistency within a data center would make it possible to provide a good user experience. We also knew that (unlike other Facebook applications), Messages was easy to federate so that a particular user could be served entirely out of a single data center making strong consistency within a single data center a critical requirement for the Messages project. Similarly, other projects, like realtime log aggregation, may be deployed entirely within one data center and are much easier to program if the system provides strong consistency guarantees.


4. Efficient random reads from disk: In spite of the widespread use of application level caches (whether embedded or via memcached), at Facebook scale, a lot of accesses miss the cache and hit the back-end storage system. MySQL is very efficient at performing random reads from disk and any new system would have to be comparable.

5. High Availability and Disaster Recovery: We need to provide a service with very high uptime to users that covers both planned and unplanned events (examples of the former being events like software upgrades and addition of hardware/capacity and the latter exemplified by failures of hardware components). We also need to be able to tolerate the loss of a data center with minimal data loss and be able to serve data out of another data center in a reasonable time frame.

6. Fault Isolation: Our long experience running large farms of MySQL databases has shown us that fault isolation is critical. Individual databases can and do go down, but only a small fraction of users are affected by any such event. Similarly, in our warehouse usage of Hadoop, individual disk failures affect only a small part of the data and the system quickly recovers from such faults.

7. Atomic read-modify-write primitives: Atomic increments and compare-and-swap APIs have been very useful in building lockless concurrent applications and are a must have from the underlying storage system.

8. Range Scans: Several applications require efficient retrieval of a set of rows in a particular range. For example all the last 100 messages for a given user or the hourly impression counts over the last 24 hours for a given advertiser.

It is also worth pointing out non-requirements:

1. Tolerance of network partitions within a single data center: Different system components are often inherently centralized. For example, MySQL servers may all be located within a few racks, and network partitions within a data center would cause major loss in serving capabilities therein. Hence every effort is made to eliminate the possibility of such events at the hardware level by having a highly redundant network design.

2. Zero Downtime in case of individual data center failure:
In our experience such failures are very rare, though not impossible. In a less than ideal world where the choice of system design boils down to the choice of compromises that are acceptable, this is one compromise that we are willing to make given the low occurrence rate of such events. We might revise this non-requirement at a later time.


3. Active-active serving capability across different data centers: As mentioned before, we were comfortable making the assumption that user data could be federated across different data centers (based ideally on user locality). Latency (when user and data locality did not match up) could be masked by using an application cache close to the user.

Some less tangible factors were also at work. Systems with existing production experience for Facebook and in-house expertise were greatly preferred. When considering open-source projects, the strength of the community was an important factor. Given the level of engineering investment in building and maintaining systems like these –– it also made sense to choose a solution that was broadly applicable (rather than adopt point solutions based on differing architecture and codebases for each workload).

After considerable research and experimentation, we chose Hadoop and HBase as the foundational storage technology for these next generation applications. The decision was based on the state of HBase at the point of evaluation as well as our confidence in addressing the features that were lacking at that point via in- house engineering. HBase already provided a highly consistent, high write-throughput key-value store. The HDFS NameNode stood out as a central point of failure, but we were confident that our HDFS team could build a highly-available NameNode (AvatarNode) in a reasonable time-frame, and this would be useful for our warehouse operations as well. Good disk read-efficiency seemed to be within striking reach (pending adding Bloom filters to HBase’’s version of LSM Trees, making local DataNode reads efficient and caching NameNode metadata). Based on our experience operating the Hive/Hadoop warehouse, we knew HDFS was stellar in tolerating and isolating faults in the disk subsystem. The failure of entire large HBase/HDFS clusters was a scenario that ran against the goal of fault-isolation, but could be considerably mitigated by storing data in smaller HBase clusters. Wide area replication projects, both in-house and within the HBase community, seemed to provide a promising path to achieving disaster recovery.

HBase is massively scalable and delivers fast random writes as well as random and streaming reads. It also provides row-level atomicity guarantees, but no native cross-row transactional support. From a data model perspective, column-orientation gives extreme flexibility in storing data and wide rows allow the creation of billions of indexed values within a single table. HBase is ideal for workloads that are write-intensive, need to maintain a large amount of data, large indices, and maintain the flexibility to scale out quickly.

Monday, March 15, 2010

Alternative to HBase + Map-Reduce

mongoDB

Combining the best features of document databases, key-value stores, and RDBMSes.

MongoDB (from "humongous") is a scalable, high-performance, open source, schema-free, document-oriented database. Written in C++, MongoDB features:

MongoDB bridges the gap between key-value stores (which are fast and highly scalable) and traditional RDBMS systems. Many companies are using MongoDB in production today.

Quick Links
Support


Thursday, December 25, 2008

Running Hadoop with HBase On CentOS Linux (Multi-Node Cluster)

1.Prerequisites

1.1 Configure single nodes

Use
my tutorial "Running Hadoop with HBase On CentOS Linux (Single-Node Cluster)"

1.2 SSH public access

Add mapping to /etc/hosts for each your nodes:

10.1.0.55 master
10.1.0.56 slave

To do that you need to add public key from master to slave ~/.ssh/authorized_keys and via verse, so eventually tou should be able to do the following:

[root@37 /usr/local/bin/hbase]ssh master
The authenticity of host 'master (10.1.0.55)' can't be established.
RSA key fingerprint is 09:e2:73:ac:6f:42:d1:da:13:20:76:10:36:29:c4:62.
Are you sure you want to continue connecting (yes/no)? yes
Warning: Permanently added 'master,10.1.0.55' (RSA) to the list of known hosts.
Last login: Thu Dec 25 07:19:16 2008 from slave
[root@37 ~]exit
logout
Connection to master closed.
[root@37 /usr/local/bin/hbase]ssh slave
The authenticity of host 'slave (10.1.0.56)' can't be established.
RSA key fingerprint is 5c:4f:81:19:ad:f3:78:02:ce:64:f1:67:10:ca:c5:b8.
Are you sure you want to continue connecting (yes/no)? yes
Warning: Permanently added 'slave,10.1.0.56' (RSA) to the list of known hosts.
Last login: Thu Dec 25 07:16:47 2008 from 38.d.de.static.xlhost.com
[root@38 ~]exit
logout
Connection to slave closed.

2.Cluster Overview

The master node will also act as a slave because we only have two machines available in our cluster but still want to spread data storage and processing to multiple machines.

Image:Hadoop-multi-node-cluster-overview.png

The master node will run the "master" daemons for each layer: namenode for the HDFS storage layer, and jobtracker for the MapReduce processing layer. Both machines will run the "slave" daemons: datanode for the HDFS layer, and tasktracker for MapReduce processing layer. Basically, the "master" daemons are responsible for coordination and management of the "slave" daemons while the latter will do the actual data storage and data processing work.

3. Hadoop core Configuration

3.1 conf/masters (master only)

The conf/masters file defines the master nodes of our multi-node cluster.

On master, update $HADOOP_HOME/conf/masters that it looks like this:


  
master

3.2 conf/slaves (master only)

This conf/slaves file lists the hosts, one per line, where the Hadoop slave daemons (datanodes and tasktrackers) will run. We want both the master box and the slave box to act as Hadoop slaves because we want both of them to store and process data.

On master, update $HADOOP_HOME/conf/slaves that it looks like this:



master
slave

If you have additional slave nodes, just add them to the conf/slaves file, one per line (do this on all machines in the cluster).


 master
slave
slave01
slave02
slave03

3.3 conf/hadoop-site.xml (all machines)

Following a sample configuration for all hosts, explanation of all parameters link




hadoop.tmp.dir
/usr/local/bin/hadoop/datastore/hadoop-${user.name}
A base for other temporary directories.



fs.default.name
hdfs://master:54310
The name of the default file system. A URI whose
scheme and authority determine the FileSystem implementation. The
uri's scheme determines the config property (fs.SCHEME.impl) naming
the FileSystem implementation class. The uri's authority is used to
determine the host, port, etc. for a filesystem.




mapred.job.tracker
master:54311
The host and port that the MapReduce job tracker runs
at. If "local", then jobs are run in-process as a single map
and reduce task.




dfs.replication
1
Default block replication.
The actual number of replications can be specified when the file is created.
The default is used if replication is not specified in create time.



mapred.reduce.tasks
8
The default number of reduce tasks per job. Typically set
to a prime close to the number of available hosts. Ignored when
mapred.job.tracker is "local".



mapred.tasktracker.reduce.tasks.maximum
8
The maximum number of reduce tasks that will be run
simultaneously by a task tracker.



mapred.child.java.opts
-Xmx500m
Java opts for the task tracker child processes.
The following symbol, if present, will be interpolated: @taskid@ is replaced
by current TaskID. Any other occurrences of '@' will go unchanged.
For example, to enable verbose gc logging to a file named for the taskid in
/tmp and to set the heap maximum to be a gigabyte, pass a 'value' of:
-Xmx1024m -verbose:gc -Xloggc:/tmp/@taskid@.gc

The configuration variable mapred.child.ulimit can be used to control the
maximum virtual memory of the child processes.




Very important: now format each node data storage before proceeding.

3.4 Start/Stop Hadoop cluster


HDFS daemons: start/stop namenode command bin/start-dfs.sh or bin/stop-dfs.sh
MapReduce daemons: run the command $HADOOP_HOME/bin/start-mapred.sh on the machine you want the jobtracker to run on. This will bring up the MapReduce cluster with the jobtracker running on the machine you ran the previous command on, and tasktrackers on the machines listed in the conf/slaves file.
[root@37 /usr/local/bin/hadoop]bin/start-dfs.sh
starting namenode, logging to /usr/local/bin/hadoop/bin/../logs/hadoop-root-namenode-37.c3.33.static.xlhost.com.out
master: starting datanode, logging to /usr/local/bin/hadoop/bin/../logs/hadoop-root-datanode-37.c3.33.static.xlhost.com.out
slave: starting datanode, logging to /usr/local/bin/hadoop/bin/../logs/hadoop-root-datanode-38.c3.33.static.xlhost.com.out
master: starting secondarynamenode, logging to /usr/local/bin/hadoop/bin/../logs/hadoop-root-secondarynamenode-37.c3.33.static.xlhost.com.out
[root@37 /usr/local/bin/hadoop]bin/start-mapred.sh
starting jobtracker, logging to /usr/local/bin/hadoop/bin/../logs/hadoop-root-jobtracker-37.c3.33.static.xlhost.com.out
master: starting tasktracker, logging to /usr/local/bin/hadoop/bin/../logs/hadoop-root-tasktracker-37.c3.33.static.xlhost.com.out
slave: starting tasktracker, logging to /usr/local/bin/hadoop/bin/../logs/hadoop-root-tasktracker-38.c3.33.static.xlhost.com.out
[root@37 /usr/local/bin/hadoop]jps]
-bash: jps]: command not found
[root@37 /usr/local/bin/hadoop]jps
28638 DataNode
29052 Jps
28527 NameNode
28764 SecondaryNameNode
28847 JobTracker
28962 TaskTracker
[root@37 /usr/local/bin/hadoop]bin/hadoop dfs -copyFromLocal LICENSE.txt testWordCount
[root@37 /usr/local/bin/hadoop]bin/hadoop dfs -ls
Found 1 items
-rw-r--r-- 1 root supergroup 11358 2008-12-25 08:20 /user/root/testWordCount
[root@37 /usr/local/bin/hadoop]bin/hadoop jar hadoop-0.18.2-examples.jar wordcount testWordCount testWordCount-out
08/12/25 08:21:46 INFO mapred.FileInputFormat: Total input paths to process : 1
08/12/25 08:21:46 INFO mapred.FileInputFormat: Total input paths to process : 1
08/12/25 08:21:47 INFO mapred.JobClient: Running job: job_200812250818_0001
08/12/25 08:21:48 INFO mapred.JobClient: map 0% reduce 0%
08/12/25 08:21:53 INFO mapred.JobClient: map 50% reduce 0%
08/12/25 08:21:55 INFO mapred.JobClient: map 100% reduce 0%
08/12/25 08:22:01 INFO mapred.JobClient: map 100% reduce 12%
08/12/25 08:22:02 INFO mapred.JobClient: map 100% reduce 25%
08/12/25 08:22:05 INFO mapred.JobClient: map 100% reduce 41%
08/12/25 08:22:09 INFO mapred.JobClient: map 100% reduce 52%
08/12/25 08:22:14 INFO mapred.JobClient: map 100% reduce 64%
08/12/25 08:22:19 INFO mapred.JobClient: map 100% reduce 66%
08/12/25 08:22:24 INFO mapred.JobClient: map 100% reduce 77%
08/12/25 08:22:29 INFO mapred.JobClient: map 100% reduce 79%
08/12/25 08:25:30 INFO mapred.JobClient: map 100% reduce 89%
08/12/25 08:26:12 INFO mapred.JobClient: Job complete: job_200812250818_0001
08/12/25 08:26:12 INFO mapred.JobClient: Counters: 17
08/12/25 08:26:12 INFO mapred.JobClient: Job Counters
08/12/25 08:26:12 INFO mapred.JobClient: Data-local map tasks=1
08/12/25 08:26:12 INFO mapred.JobClient: Launched reduce tasks=9
08/12/25 08:26:12 INFO mapred.JobClient: Launched map tasks=2
08/12/25 08:26:12 INFO mapred.JobClient: Rack-local map tasks=1
08/12/25 08:26:12 INFO mapred.JobClient: Map-Reduce Framework
08/12/25 08:26:12 INFO mapred.JobClient: Map output records=1581
08/12/25 08:26:12 INFO mapred.JobClient: Reduce input records=593
08/12/25 08:26:12 INFO mapred.JobClient: Map output bytes=16546
08/12/25 08:26:12 INFO mapred.JobClient: Map input records=202
08/12/25 08:26:12 INFO mapred.JobClient: Combine output records=1292
08/12/25 08:26:12 INFO mapred.JobClient: Map input bytes=11358
08/12/25 08:26:12 INFO mapred.JobClient: Combine input records=2280
08/12/25 08:26:12 INFO mapred.JobClient: Reduce input groups=593
08/12/25 08:26:12 INFO mapred.JobClient: Reduce output records=593
08/12/25 08:26:12 INFO mapred.JobClient: File Systems
08/12/25 08:26:12 INFO mapred.JobClient: HDFS bytes written=6117
08/12/25 08:26:12 INFO mapred.JobClient: Local bytes written=19010
08/12/25 08:26:12 INFO mapred.JobClient: HDFS bytes read=13872
08/12/25 08:26:12 INFO mapred.JobClient: Local bytes read=8620

4. HBase configuration

4.1 $HBASE_HOME/hbase-site.sh



hbase.rootdir
hdfs://master:54310/hbase
The directory shared by region servers.



hbase.master
master:60000
The host and port that the HBase master runs at.




4.2 $HBASE_HOME/conf/regionservers

master
slave

4.3 Start HBase on master

Use command $HBASE_HOME/bin/hbase-start.sh

[root@37 /usr/local/bin/hbase]bin/start-hbase.sh
starting master, logging to /usr/local/bin/hbase/bin/../logs/hbase-root-master-37.c3.33.static.xlhost.com.out
slave: starting regionserver, logging to /usr/local/bin/hbase/bin/../logs/hbase-root-regionserver-38.c3.33.static.xlhost.com.out
master: starting regionserver, logging to /usr/local/bin/hbase/bin/../logs/hbase-root-regionserver-37.c3.33.static.xlhost.com.out
[root@37 /usr/local/bin/hbase]jps
30362 SecondaryNameNode
30455 JobTracker
30570 TaskTracker
30231 DataNode
32003 Jps
31772 HMaster
30115 NameNode
31919 HRegionServer

On slave you suppose to see the following output:

[root@38 /usr/local/bin/hbase]jps
513 DataNode
6360 Jps
6214 HRegionServer
628 TaskTracker

Thursday, December 11, 2008

Distributed Systems and Web Scalability Resources

Distributed Systems and Web Scalability Resources

No long write-ups this week, just a short list of some great resources that I've found very inspirational and thought provoking. I've broken these resources up into two lists: Blogs and Presentations.

Blogs

The blogs listed below are ones that I subscribe to and are filled with some great posts about capacity planning, scalability problems and solutions, and distributed system information. Each blog is authored by exceptionally smart people and many of them have significant experience building production-level scalable systems.

Nati Shalom's Blog: Discussions about middleware and distributed technologies
http://natishalom.typepad.com/nati_shaloms_blog/

All Things Distributed: Werner Vogels' weblog on building scalable and robust distributed systems.
http://www.allthingsdistributed.com/

High Scalability: Building bigger, faster, more reliable websites
http://highscalability.com/

ProductionScale: Information Technology, Scalability, Technology Operations, and Cloud Computing
http://www.productionscale.com/

iamcal.com
http://www.iamcal.com/ (the "talks" section is particularly interesting)

Kitchen Soap: Thoughts on capacity planning and web operations
http://www.kitchensoap.com/

MySQL Performance Blog: Everything about MySQL Performance
http://www.mysqlperformanceblog.com/

Presentations

The presentations listed below are from the SlideShare site and are primarily the slides used to accompany scalability talks from around the world. Many of them outline the problems that various companies have encountered during their non-linear growth phases and how they've solved them by scaling their systems.

Scalable Internet Architectures
http://www.slideshare.net/shiflett/scalable-internet-architectures

How to build the Web
http://www.slideshare.net/simon/how-to-build-the-web

Netlog: What we learned about scalability & high availability
http://www.slideshare.net/folke/netlog-what-we-learned-about-scalability-high-availability-430211

Database Sharding at Netlog
http://www.slideshare.net/oemebamo/database-sharding-at-netlog-presentation

MySQL 2007 Techn At Digg V3
http://www.slideshare.net/epee/mysql-2007-tech-at-digg-v3

Flickr and PHP
http://www.slideshare.net/coolpics/flickr-44054

Scalable Web Architectures: Common Patterns and Approaches
http://www.slideshare.net/techdude/scalable-web-architectures-common-patterns-and-approaches

How to scale your web app
http://www.slideshare.net/Georgio_1999/how-to-scale-your-web-app

Google Cluster Innards
http://www.slideshare.net/ultradvorka/google-cluster-innards

Sharding Architectures
http://www.slideshare.net/guest0e6d5e/sharding-architectures

Amazon EC2 setup - link
Yahoo Hadoop tutorial - link
Michael Noll blog - link
Hadoop main page - link
Google lectures - link
HBase resources - link
Distributed computing(IBM) - link
HBase and BigTable - link




Understanding HBase column-family performance options - link

Debugging and Tuning Map-Reduce Applications


by Arun C Murthy, Principal Engineer at Yahoo! and Member of Apache Hadoop PMC

http://www.vimeo.com/2085477