Hadoop入门 卢学裕@优酷网 2012.07.

Presentation on theme: "Hadoop入门 卢学裕@优酷网 2012.07."— Presentation transcript:

1 Hadoop入门

2 Outlines Hadoop Overview HDFS Map-reduce Programming Paradigm
Hadoop Map-reduce Job Scheduler Resources

3 Hadoop, Why? 数据太多了，需要能存储、快速分析Pb级数据集的系统 单机的存储、IO、内存、CPU有限，需要可扩展的集群
使用门槛低，数据分析是个庞杂的问题，MPI太复杂 单点故障问题 –机器多了单点故障成为正常的异常 –节点有增有减 Hadoop就是一个满足易用性、可靠性、可扩展性的存储计算平台，还是开源的！

4 Hadoop History Dec 2004 – Google GFS paper published
July 2005 – Nutch uses MapReduce Feb 2006 – Becomes Lucene subproject Apr 2007 – Yahoo! on 1000-node cluster Apr 2008 – Fastest QuickSort on 1TB Jan 2008 – An Apache Top Level Project Jul 2008 – A 4000 node test cluster

5 Hadoop-related projects
Hive：数据仓库，facebook贡献 PIG：并行计算的一种高级语言，yahoo贡献 Nutch：网页搜索软件，不只是爬虫 Avro：数据序列化系统 Chukwa：用于管理大规模分布式集群的数据收集系统 ZooKeeper：用于分布式应用的高性能协同服务 Hbase：类似于BigTable的，key-value数据库系统 Mahout：分布式机器学习和数据挖掘的Lib Hama：基于BSP的超大规模科学计算框架 ……

6 Who Uses Hadoop Amazon 著名的AWS Adobe
Adknowledge： behavioral targeting, clickstream analytics Alibaba Baidu: 搜索日志分析；每周处理3000TB 数据 Bixo Labs：web mining Datagraph：处理RDF数据，存储、索引 EBay ： 532 nodes，搜索优化和研究 ETH Zurich Systems Group：教学《 Massively Parallel Data Analysis with MapReduce 》 Facebook ： 1100 nodes, 12PB; 300 nodes, 3PB FOX：3个Cluster 用于日志分析、数据挖掘、机器学习 Freestylers：构建基于图片的推荐系统 Google Gruter. Corp：索引、Link分析、数据挖掘 Hulu： Hbase hosting IBM Krugle ：源代码搜索 Last.fm：图表计算、A/B测试，user profile分析，cookie级的报表处理 Lineberger Comprehensive Cancer Center：癌症相关的研究，使用SeqWare LinkedIn ：这人你可能认识 The New York Times PARC：分析Wikipedia里的冲突 Pressflip：个性化搜索，训练SVM模型 Yahoo！： 4000 nodes (2*4cpu boxes w 4*1TB disk & 16GB RAM) 优酷土豆 More on

7 Goals of HDFS 大数据集存储 – 10K nodes, 100 million files, 10 PB 应付硬件故障
– 用文件多副本应付故障问题 – 故障自动检测和恢复 更适合批量处理 – 搬移计算比搬移数据更廉价 – 数据一次写入，多次读取 – 更注重数据读取的高吞吐量，而不是低延时 适应复杂的硬件及软件平台

8 The File System 一个集群只有一个Namespace
跟Unix的文件系统Namespace很相似，不过不支持Hard link、Soft link 文件分块存储 一般一块大小为64M，可配置 每块会被复制在多个DataNode上 支持回收站 当一个文件被删除时会先放入用户下的回收站 回收站会被定期清除 恢复的方式是将文件从回收站移出

9

10 NameNode Metadata Meta-data 存在内存中 – 整个Meta-data放入主内存
– No demand paging of meta-data Meta-data记录了 – 文件列表信息 – 每个文件的块列表 – 每个块对应的DataNode – 文件属性，如创建时间、创建者、几份副本等 Transaction Log (EditLog ) –记录了文件系统的每个变化，如创建文件、删除文件、修改文件的副本数等 – EditLog会被合并为FsImage并存入磁盘 Meta-data 磁盘故障 – NameNode可以维护多份数据

11 DataNode A Block Server – 将数据存储本机文件系统(e.g. ext3)
– 存储数据块的Meta-data(e.g. CRC, ID) 汇报 – 启动时向NameNode注册本地存储的文件块 –定期向NameNode报告本机存活（心跳） 数据输送 – 接收来自客户端的写数据 – 向客户端发送数据 –将数据传输到指定的 DataNodes

12 Block Replica Placement
机架感知 NameNode能感知机架，选择较优的方式 假设有3份或以上，目前的策略是 一份放在本地节点上 第二份放在另外一个机架的节点上 第三份放在跟第二份同机架的不同节点上 其他的随机放置 客户端从最近的块读取

13 Data Correctness – Use CRC32 – 由客户端负责计算CRC –客户端从DataNode读取数据和checksum
文件写入 – 由客户端负责计算CRC – checksum存放在DataNode 文件读取 –客户端从DataNode读取数据和checksum –由客户端校验，如果不通过，则客户端尝试从其他的副本读取数据

14 FS Shell FS DFSAdmin fsck Balancer hadoop fs -mkdir /foodir
hadoop fs -rmr /foodir hadoop fs -cat /foodir/myfile.txt hadoop fs -tail /foodir/myfile.txt Chmod，chown，put，mv，cp，du，dus 更多命令请运行 hadoop fs –help获取 DFSAdmin Safemode，upgradeProgress，refreshNodes，… fsck 文件系统检查 Balancer 集群均衡

15 Web UI

16 Map-reduce Programming Paradigm
input | map | shuffle | reduce | output 最简单的实现方式 cat * | grep ‘java’| sort | uniq -c | cat > file 实现这种编程范式的有 Google Hadoop Oracle Teradata ……

17 Hadoop Map/Reduce (input) <k1, v1> -> map -> <k2, v2> -> combine* -> <k2, v2> -> reduce -> <k3, v3> (output) combine 过程可能没有，也可能有多次

18 WordCount (©周敏@Taobao)
the 1 weather 1 is 1 good 1 a 1 The weather is good good 1 a 1 good 4 today 1 is 1 good 1 Today is good guy 1 is 1 guy 1 is 4 man 2 the 1 this 1 guy 1 is 1 a 1 man 1 This guy is a man man 1 the 1 good 1 man 1 is 1 Good man is good this 1 this 1 today 1 weather 1 today 1 weather 1

19 WordCount Mapper public static class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> { private final static IntWritable ONE= new IntWritable(1); private Text word = new Text(); protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { StringTokenizer itr = new StringTokenizer(value.toString()); while (itr.hasMoreTokens()) { word.set(itr.nextToken()); context.write(word, ONE); }

20 WordCount Reducer public static class IntSumReducer extends Reducer<Text,IntWritable,Text,IntWritable> { private IntWritable result = new IntWritable(); public void reduce(Text key, Iterable<IntWritable> values, Context context ) throws IOException, InterruptedException { int sum = 0; for (IntWritable val : values) { sum += val.get(); } result.set(sum); context.write(key, result);

21 WordCount Job Setup public static void main(String[] args) throws Exception { String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs(); if (otherArgs.length != 2) { System.err.println("Usage: wordcount <in> <out>"); System.exit(2); } Job job = new Job(new Configuration(), "word count"); job.setJarByClass(WordCount.class); //设置输入 job.setInputFormatClass(TextInputFormat.class); FileInputFormat.addInputPath(job, new Path(otherArgs[0])); //Map job.setMapperClass(WordCountMapper.class); //Combine job.setCombinerClass(IntSumReducer.class); //Reduce job.setReducerClass(IntSumReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); //设置输出 FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); //提交Job并执行 System.exit(job.waitForCompletion(true) ? 0 : 1);

22 Inject Points Input Map Combine* Shuffling Sort Grouping Reduce Output
job.setInputFormatClass() Map job.setMapperClass() Combine* job.setCombinerClass() Shuffling job.setPartitionerClass() Sort job.setSortComparatorClass() Grouping job.setGroupingComparatorClass() Reduce job.setReducerClass() Output job.setOutputFormatClass()

23 Job Tracker & Task Tracker
Job & Task 一个Job会被分成多个Task执行 一个Task对应一个Map或者Reduce Job Tracker 运行在Master上 ，管理和跟踪每个Job 收集Task的信息状态，并汇总 重新调度失败的任务 Task Tracker 向Job Tracker汇报状态（心跳） 运行在每个计算节点上，管理和跟踪每个Task 收集task的信息，并提供给Job Tracker

24 Job Sheduler FIFO 先到先得，排队执行 Fair Scheduler（公平调度器）
它的目的是让所有的作业随着时间的推移，都能平均的获取等同的共享资源。 按资源池（pool）来组织作业，并把资源按配置分到这些资源池里 Capacity Scheduler（容量调度器） 支持多个queue，每个Job提交到一个queue里 支持内存调度，对于需要高内存的任务，调度到有足够内存的节点

25 Job Scheduler Web UI

26 Job Shell hadoop job –kill <job-id> hadoop job –list
hadoop job -set-priority <job-id> <priority> hadoop job -status <job-id> hadoop job -kill-task <task-id> hadoop job -fail-task <task-id>

27 Job Web UI

28 Select COUNT(DISTINCT guid) as uv from youku_pv;
Hive Select COUNT(DISTINCT guid) as uv from youku_pv;

29 Hadoop Next Generation
NameNode单点故障 YARN Resource Manager Job Scheduling/monitoring Application Submission Client ApplicationMaster New Programming Paradigm MPI Master-Worker Iterative models “Customize” by yourself

30 Hadoop Next Generation Architecture

31 优酷数据平台V2 分布式实时计算系统 网站 反作弊系统 分布式日志收集系统 广告 分布式实时数据流处理系统 无线 HADOOP HDFS
…… 接口系统监控 作弊监测 广告计数 PlayLog记录 播放计数 个性化推荐 计数/监控 准实时计算 网站 反作弊系统 分布式日志收集系统 广告 分布式实时数据流处理系统 无线 HADOOP HDFS HBASE HIVE 搜索 MAP-REDUCE 统计分析 数据挖掘 运营分析 相关推荐 用户分析 个性化推荐 数据开放平台 精准广告 …… 播放器 调度系统 大数据批量计算系统

32 优酷推荐系统

33 指数&VideoProfile 演示指数

34 Resources Hadoop Hadoop World 推荐资料 http://hadoop.apache.org/
Hadoop World 推荐资料 《Hadoop：The.Definitive.Guid 3rd 》 《Hadoop实战》 《Hadoop权威指南》 《hadoop开发者》杂志

35 Thanks