发表于 | 分类于
字数统计: | 阅读时长 ≈

分布式存储引擎层包含三个模块: RootServer,UpdateServer,ChunkServer

  • RootServer: 用于整体控制,实现子表分布,副本复制,负载均衡,机器管理以及Schema管理
  • UpdateServer: 用于存储增量数据
  • ChunkServer: 用于存储基线数据

发表于 | 分类于
字数统计: | 阅读时长 ≈

OceanBase

支持强一致性和跨行跨表事务,支持服务器线性扩展

OceanBase整体架构图

定期合并与数据分发:

  1. UpdateServer冻结当前的活跃内存表,生成冻结内存表,并开启新的活跃内存表,后续的更新操作都写入到新的活跃内存表
  2. UpdateServer通知RootServer数据版本发生了变化,之后RootServer通过心跳消息通知ChunkServer
  3. 每台ChunkServer启动定期合并或者数据分发操作,从UpdateServer获取每个子表对应的增量更新数据

查询结果 = 旧子表+ 冻结内存表+新的活跃内存表
= 新子表+ 新的活跃内存表

如何实现强一致性:
1. 操作日志同步到主备的情况下才能够返回客户端写入成功
2. 为了提高可用性,如果主机往备机同步操作日志失败,比如备机故障,主机可以将备机从同步列表中剔除,本地跟新成功后就返回客户端写入成功.

OceanBase数据分为基线数据和增量数据两部分,基线数据分布在多台ChunkServer上,增量数据全部放在一台UpdateServer上.

读写事务

用户的读写请求,都发给MergeServer,

  • 只读事务: 由MergeServer发送给相应的ChunkServer分别执行后再合并每个ChunkServer的执行结果
  • 读写事务: 由MergeServer进行预处理后,再发生给UpdateServer执行.

数据正确性

  • 数据存储校验: 每个存储记录同时保存64位CRC校验码.数据被访问时,重新计算和比对校验码
  • 数据传输校验: 每个传输记录同时传输64位crc校验码,数据被接收后,重新计算和比对校验码
  • 数据镜像校验: UpdateServer的内存表在主备间必须保持一致,为此UpdateServer为每个MemTable生成一个校验码,MemTable 每次更新时,校验码同步更新并记录在对应的操作日志中,备UpdateServer收到操作日志并重放到MemTable时,也同步更新MemTable检验码并与接收到的校验码对照.
  • 数据副本校验: ChunkServer在定期合并生成新的子表时,也同时为每个子表生成一个校验码,并碎新子表汇报给RootServer,以便RootServer核对同一个子表不同副本的校验码

发表于 | 分类于
字数统计: | 阅读时长 ≈

分布式系统

Paxos选举协议以及两阶段提交

基本概念

异常

异常类型

  • 服务器宕机

  • 网络异常

  • 磁盘故障

    超时

  • 接口幂等性,覆盖写是一种常见的幂等操作

    一致性

  • 副本是分布式存储系统容错技术的唯一手段

  • 如何保证副本之间的一致性是整个分布式系统的理论核心

理解一致性可以从两个角度:用户/存储系统

  • 客户端:

    • 强一致
    • 弱一致
    • 最终一致性: 最终一致有一个“不一致窗口”,不一致窗口大小以来以下几个因素:交互延迟,系统负载,以及复制协议要求同步的副本数
      最终一致性描述比较粗略,其他常见的变体如下:

  • 读写(Read-your-writes)一致性: 如果客户端A写入了最新的值,那么A的后续操作都会读取到最新的值,但是其他用户可能要过一会儿才能看到

  • 会话(session)一致性: 要求客户端和存储系统交互的整个会话期间保证读写一致性,如果原有会话因为某种原因失效,会话之间的操作要回滚掉

  • 单调读一致性: 如果客户端A已经读取了对象的某个值,那么后续操作将不会读取到更早的值

  • 单调写一致性: 客户端A的写操作按顺序我拿成,这意味着对与同一个客户端的操作,存储系统的多副本也要按照与客户端相同的顺序完成

  • 从存储系统角度看:

    • 副本一致性:
    • 更新顺序一致性

数据分布

分布式系统区别于传统单机系统在于能够数据分布到多个节点,并在多个节点之间实现负载均衡.
数据分布方式主要有两种:
- 哈希分布,如一致性哈希
- 顺序分布

哈希分布

衡量机器负载设计很多因素:如机器load,cpu,内存,磁盘以及网络等资源,分布式存储需要能够自动识别负载高的节点,当某台机器负载高时,实现自动迁移

传统哈希分布,当服务器上下线时,N值发生变化,数据映射完全被打乱.几乎所有数据需要重新分布,这将带来大量的数据迁移

一致性哈希有点在于节点加入/删除是只会影响到在哈希环中相邻的节点,而对其他节点没影响

顺序分布

将大表顺序划分为连续的范围,每个范围称为一个子表,总控服务器负责将这些子表按照一定的策略分配到存储节点上.

负载均衡

  • 分布式系统每个集群中一般都有个总恐怖节点,其他节点为工作节点,工作节点通过心跳包将节点负载信息(如cpu,内存,磁盘,网络等信息)发送到主控节点.主控节点计算出工作节点的负载以及需要迁移的数据,生成迁移任务放入迁移队列中等待执行.
  • 负载均衡操作需要控制节奏,因为在迁移过程中机器服务能力会下降,一般新加机器迁移过程会持续300分钟到1个小时
  • 分布式存储一版是多副本,主副本对外提供服务,迁移备副本不会对服务造成影响,迁移主副本可以先将数据的读写服务切换到其他备副本,再迁移

复制

主副本将写请求复制到其他备副本,最常见的做法是同步操作日志,主副本确定操作的顺序并写入到日志中复制给其他副本

复制协议分为两种:

  • 强同步协议要求: 主备同步完成后才能返回客户端写成功.假设所有副本的个数为N,且N>2, 那么实现强同步协议时,主副本只要并发地发给所有备副本并等待回复,只要至少1个备副本返回成功就可以回复客户端成功.
  • 异步复制: 主副本不需要等待备副本的回应,只需要在本地修改成功就可以告知客户端写操作成功

基于复制协议,要求任何时刻只能有一个副本为主副本,由它来确定写操作之间的顺序.如果主副本故障,需要选举一个备副本成为新的主副本,这步操作称为选举,经典的选举协议为(Paxos协议)

一致性与可用性

CAP理论: 一致性(Consistency),可用性(Availability)以及分区可容忍性(Tolerance of network Partition)三者不能同时满足.

  • 一致性: 读操作总能读取到之前完成的写操作结果,满足这个条件的系统称为强一致性系统
  • 可用性: 读写操作在单台机器发生故障情况下仍然能正常执行
  • 分区可容忍性: 机器故障,网络故障情况下仍能满足一致性和可用性

最大可用性模型: 正常情况下强同步模式,异常情况切换为异步复制模式

容错

故障检测往往通过租约协议

故障检测

故障检测最自然想到的是心跳机制,但还是存在问题(没有收到心跳包并不一定能100分确认发生故障并停止了服务)

传统心跳机制存在问题在于: 机器A和机器B之间需要对“机器B是否应为被认为发生故障且停止服务”达成一致.

租约机制就是带有超时时间的一种授权,假设机器A需要检测机器B是否发生故障,机器A可以给机器B发放租约,机器B持有的租约的有效期范围内才允许提供服务,否则主动停止复制.机器B的租约快到期时向A重新申请租约来延长有效期.

增加提前量,假设B机器租约有效期是10s,A机器必须啊哟加一个提前量,比如11s,才能任务机器B的租约过期.

故障恢复

常见分布式存储系统分为两种结构:单层结构/双层结构.

可扩展性

通过数据分布,复制以及容错等机制,能够将分布式存储系统部署到成千上万台服务器

总控节点

总控节点一般用于维护数据分布信息,执行工作机管理,数据定位,故障检测与恢复,负载均衡等全局调度工作,那么总控节点是否会成为性能瓶颈呢?

不太可能,而且可以通过一些手段来避免

分布式协议:

两阶段提交协议:
A组织B,C,D三人去爬长城,如果所有人同意去爬长城,那么活动将举行,如果有一个人不同意去爬长城,则活动取消.

但是

两阶段提交协议是阻塞协议,执行过程需要锁住其他更新,且不能容错

google spanner

https://segmentfault.com/a/1190000009707788

发表于 | 分类于
字数统计: | 阅读时长 ≈

概述

分布式文件系统

分布式键值系统

分布式表格系统

分布式数据库

单机存储系统

单机存储引擎就是hash表,b树等数据结构在机械磁盘,ssd等持久化上的实现

硬件基础

cpu架构

io总线

Numa架构

Numa架构

Interl X48主板南北桥架构

网络拓扑

性能参数

硬件性能对比

15000 转的stata盘顺序读取带宽可达到100MB以上,由于磁盘寻到的时间大约是10ms,顺序读取1Mb数据时间为: 磁盘寻道时间+ 数据读取时间,即 10ms+ 1MB/100MB/s X1000 = 20ms

存储介质对比

存储层次架构

热数据(访问频繁)存储到ssd中,冷数据(访问不频繁的)存储到磁盘中

单机存储引擎

哈希存储引擎

哈希表的持久化实现,支持增删读改操作,以及随机读取操作,但不支持顺序扫描,对应的存储系统为键值(key-Value)存储系统

B树存储引擎

B树的持久化实现,不仅支持单条记录的增删读改操作,还支持顺序扫描,对应的存储系统是关系数据库,当然键值系统也可以通过B树存储引擎实现

LSM树

LSM树存储引擎和b树存储引擎一样,支持增、删、改、随机读以及顺序扫描.它通过批量转储技术规避磁盘随机写入问题

发表于 | 分类于
字数统计: | 阅读时长 ≈

6.824 Lab 1: MapReduce

Part1 Map/Reduce input and output

主要是实现 mapreduce/common_map.go 中的 doMap() 函数以及 mapreduce/common_reduce.go 中的 doReduce() 函数

doMap()函数:

  1. 读取 inFile 文件的内容，然后经过用户定义的 mapF() 函数得到一个 KeyValue列表
  2. 对于每个KeyValue值，经过ihash平均分配给nReduce个中间文件,中间文件可由reduceName() 得到名字，注意读写文件需要用json来序列化

疑问:

  • 如果文件特别大,不可能一次性全读到内存中啊
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
func doMap(
	jobName string, // the name of the MapReduce job
	mapTask int,    // which map task this is
	inFile string,
	nReduce int, // the number of reduce task that will be run ("R" in the paper)
	mapF func(filename string, contents string) []KeyValue,
) {
	// 读取文件，并处理产生键值对
	cont, _ := ioutil.ReadFile(inFile)
	contents := string(cont)
	resultKV := mapF(inFile, contents)

	reduceFiles := make([]*os.File, nReduce)
	for i := 0; i < nReduce; i++ {
		file, _ := os.OpenFile(reduceName(jobName, mapTask, i), os.O_CREATE|os.O_WRONLY, 0666)
		reduceFiles[i] = file
		defer file.Close()
	}

	// 将键值对写入nReduce个文件中
	for _, keyValue := range resultKV {
		r := ihash(keyValue.Key) % nReduce
		file := reduceFiles[r]
		enc := json.NewEncoder(file)
		_ = enc.Encode(&keyValue)
	}
}

doReduce

  1. 从所有中间文件中读取到内存中,读出KeyValue值后将相同key的value合并，这里可以使用map。然后按照key进行排序
  2. 对新的key-value使用 reduceF() 函数，即可得到例如 hello: 2, word: 3 的词频结果，写入新的merge文件,merge文件可由mergeName()得到名字
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
func doReduce(
	jobName string, // the name of the whole MapReduce job
	reduceTask int, // which reduce task this is
	outFile string, // write the output here
	nMap int,       // the number of map tasks that were run ("M" in the paper)
	reduceF func(key string, values []string) string,
) {
	var keyValues []KeyValue
	for i := 0; i < nMap; i++ {
		reduceFileName := reduceName(jobName, i, reduceTask)
		file, err := os.Open(reduceFileName)
		if err != nil {
			fmt.Println("doReduce can not open:", reduceFileName)
			return
		}
		defer file.Close()
		br := bufio.NewReader(file)
		for {
			line, _, next := br.ReadLine()
			var tmp KeyValue
			json.Unmarshal(line, &tmp)
			if next == io.EOF {
				break
			}
			keyValues = append(keyValues, tmp)
		}
	}
	sort.Slice(keyValues, func(i, j int) bool {
		return keyValues[i].Key < keyValues[j].Key
	})

	oFile, _ := os.OpenFile(outFile, os.O_CREATE|os.O_WRONLY, 0666)
	defer oFile.Close()
	enc := json.NewEncoder(oFile)

	tmp := keyValues[0].Key
	var values []string
	for _, keyValue := range keyValues {
		if keyValue.Key == tmp {
			values = append(values, keyValue.Value)
		} else {
			enc.Encode(KeyValue{tmp, reduceF(tmp, values)})
			values = []string{keyValue.Value}
			tmp = keyValue.Key
		}
	}
	enc.Encode(KeyValue{tmp, reduceF(tmp, values)})
}

最后在终端输入 go test -run Sequential进行测试，如果出现

1
2
3
4
5
6
7
$ cd 6.824
$ export "GOPATH=$PWD"  # go needs $GOPATH to be set to the project's working directory
$ cd "$GOPATH/src/mapreduce"
$ go test -run Sequential

.\master_rpc.go:48: debug call has arguments but no formatting directives
FAIL 6.824/src/mapreduce [build failed]

注释掉 master_rpc.go的48行 debug("RegistrationServer: accept error", err) 即可

Part2 Single-worker word count

实现main/wc.go 中的mapF() and reduceF() 方法,统计单词出现次数

1
2
3
4
5
6
7
8
9
10
11
func mapF(filename string, contents string) []mapreduce.KeyValue {
	// Your code here (Part II).
	words := strings.FieldsFunc(contents, func(r rune) bool {
		return !unicode.IsLetter(r)
	})
	var kvs []mapreduce.KeyValue
	for _, word := range words {
		kvs = append(kvs, mapreduce.KeyValue{Key: word, Value: "1"})
	}
	return kvs
}
1
2
3
4
func reduceF(key string, values []string) string {
	// Your code here (Part II).
	return strconv.Itoa(len(values))
}

运行:

1
2
3
4
$ cd 6.824
$ export "GOPATH=$PWD"
$ cd "$GOPATH/src/main"
$ go run wc.go master sequential pg-*.txt

Part III: Distributing MapReduce tasks

文案翻译:

您当前的实现只能一次运行一个任务。 Map / Reduce最大的卖点之一是，它可以自动并行化普通的顺序代码，而无需开发人员进行任何额外的工作。 在本部分的实验中，您将完成一个MapReduce版本，该版本将工作划分为一组在多个内核上并行运行的工作线程。 虽然不像实际的Map / Reduce部署那样分布在多台计算机上，但是您的实现将使用RPC来模拟分布式计算。

mapreduce / master.go中的代码完成了管理MapReduce作业的大部分工作。 我们还在mapreduce / worker.go中为您提供了工作线程的完整代码，并在mapreduce / common_rpc.go中为您提供了处理RPC的一些代码。

您的工作是在mapreduce / schedule.go中实现schedule（）。 Master在MapReduce作业期间两次调用schedule（），一次在Map阶段，一次在Reduce阶段。 schedule（)的工作是将任务分发给可用的worker。 通常tasks数超过worker数，因此schedule（）必须给每个工作线程一系列任务， schedule（）应等待所有任务完成，然后返回。

schedule（）从registerChan中获取空闲的worker的地址，然后将任务非配给空闲的worker执行.

schedule（）通过Worker.DoTaskRPC发送给worder来通知worker执行任务。 该RPC的参数由DoTaskArgs在mapreduce / common_rpc.go中定义。 File元素仅由Map任务使用，是要读取的文件的名称； schedule（）可以在mapFiles中找到这些文件名。

使用mapreduce / common_rpc.go中的call（）函数将RPC发送给工作程序。 第一个参数是worker的地址，从registerChan读取。 第二个参数应为Worker.DoTask。 第三个参数应为DoTaskArgs结构，最后一个参数应为nil。

使用go test -run TestParallel来测试您的解决方案。 这将执行两个测试，TestParallelBasic和TestParallelCheck； 后者验证您的scheduler程序执行task的并行度

思路:

  • 使用for循环启动需要执行的任务
  • 使用goroutine执行每个任务
  • 等待所有任务执行结束
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
func schedule(jobName string, mapFiles []string, nReduce int, phase jobPhase, registerChan chan string) {
	var ntasks int
	var n_other int // number of inputs (for reduce) or outputs (for map)
	switch phase {
	case mapPhase:
		ntasks = len(mapFiles)
		n_other = nReduce
	case reducePhase:
		ntasks = nReduce
		n_other = len(mapFiles)
	}

	fmt.Printf("Schedule: %v %v tasks (%d I/Os)\n", ntasks, phase, n_other)

	// All ntasks tasks have to be scheduled on workers. Once all tasks
	// have completed successfully, schedule() should return.
	//
	// Your code here (Part III, Part IV).
	var waitGroup sync.WaitGroup
	for i := 0; i < ntasks; i++ {
		waitGroup.Add(1)
		var taskArgs DoTaskArgs
		taskArgs.JobName = jobName
		taskArgs.Phase = phase
		taskArgs.TaskNumber = i
		taskArgs.NumOtherPhase = n_other
		if phase == mapPhase {
			taskArgs.File = mapFiles[i]
		}
		go func() {
			defer waitGroup.Done()
			worker := <-registerChan
			if !call(worker, "Worker.DoTask", &taskArgs, nil) {
				log.Fatal("call RPC failed")
			}
			go func() { registerChan <- worker }()
		}()
	}
	waitGroup.Wait()
	fmt.Printf("Schedule: %v done\n", phase)
}

参考:
Go官方库RPC开发指南
Package rpc
谈谈 Golang 中的 Data Race
sync.Mutex 互斥锁
Golang中WaitGroup使用的一点坑

Part IV: Handling worker failures

schedule调度的任务可能会发生错误，需要将发生错误的任务在一个空闲的worker上重新执行。这里可以对发生错误的任务使用递归的方式在一个新的worker执行。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
func schedule(jobName string, mapFiles []string, nReduce int, phase jobPhase, registerChan chan string) {
	var ntasks int
	var n_other int // number of inputs (for reduce) or outputs (for map)
	switch phase {
	case mapPhase:
		ntasks = len(mapFiles)
		n_other = nReduce
	case reducePhase:
		ntasks = nReduce
		n_other = len(mapFiles)
	}

	fmt.Printf("Schedule: %v %v tasks (%d I/Os)\n", ntasks, phase, n_other)

	// All ntasks tasks have to be scheduled on workers. Once all tasks
	// have completed successfully, schedule() should return.
	//
	// Your code here (Part III, Part IV).
	var waitGroup sync.WaitGroup
	for i := 0; i < ntasks; i++ {
		waitGroup.Add(1)
		var taskArgs DoTaskArgs
		taskArgs.JobName = jobName
		taskArgs.Phase = phase
		taskArgs.TaskNumber = i
		taskArgs.NumOtherPhase = n_other
		if phase == mapPhase {
			taskArgs.File = mapFiles[i]
		}
		go func() {
			defer waitGroup.Done()
			for {
				worker := <-registerChan
				if call(worker, "Worker.DoTask", taskArgs, nil) {
					go func() {
						registerChan <- worker
					}()
					break
				}

			}

		}()
	}
	waitGroup.Wait()
	fmt.Printf("Schedule: %v done\n", phase)
}

Part V: Inverted index generation (optional, does not count in grade)

倒排索引

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
func mapF(document string, value string) (res []mapreduce.KeyValue) {
	// Your code here (Part V).
	words := strings.FieldsFunc(value, func(r rune) bool {
		return !unicode.IsLetter(r)
	})
	DocMaps := make(map[string]string)
	var kv []mapreduce.KeyValue
	for _, word := range words {
		DocMaps[word] = document
	}
	for k, doc := range DocMaps {
		kv = append(kv, mapreduce.KeyValue{Key: k, Value: doc})
	}
	return kv
}

// The reduce function is called once for each key generated by Map, with a
// list of that key's string value (merged across all inputs). The return value
// should be a single output value for that key.
func reduceF(key string, values []string) string {
	// Your code here (Part V).
	nDoc := len(values)
	sort.Strings(values)
	resString := strconv.Itoa(nDoc) + " "
	for _, v := range values {
		resString = resString + v + ","
	}
	return resString[:(len(resString)-1)]
}

参考

用go实现MapReduce部分功能
MIT 6.824 Distributed Systems Lab-1 笔记

发表于 | 分类于
字数统计: | 阅读时长 ≈

什么是rdd

RDD，全称为 Resilient Distributed Datasets,弹性分部署数据集,是为了对用户操作的简化,以面向对象的方式提供了RDD很多方法,rdd是懒执行的,分为转换和行动两部分

rdd特性

  • 不可变,所有rdd操作会产生一个新的rdd
  • 可分区,通过将数据进行分区保存
  • 弹性
    存储的弹性: 内存与磁盘的自动切换
    容错的弹性: 数据丢失可以自动恢复
    计算的弹性: 计算出错重试机制
    分片的弹性: 根据需要重新分片

spark都做了什么:

rdd创建

rdd创建方式分为3种:

从集合中创建
在Driver（驱动程序）中一个已经存在的集合（数组）上创建，SparkContext对象代表到Spark集群的连接，可以用来创建RDD、广播变量和累加器。可以复制集合的对象创建一个支持并行操作的分布式数据集（ParallelCollectionRDD）。一旦该RDD创建完成，分布数据集可以支持并行操作，比如在该集合上调用Reduce将数组的元素相加。

1
2
3
4
scala> sc.makeRDD(0 to 10)
res0: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at makeRDD at <console>:25
scala> res0.collect
res1: Array[Int] = Array(0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10)

Apache Spark RDD 论文中文版 | ApacheCN
Spark RDD 论文简析

发表于 | 分类于
字数统计: | 阅读时长 ≈

Zookeeper 集群搭建

环境: CentOS release 6.10 + zookeeper-3.4.14

环境准备

特别注意防火墙问题

环境准备

下载解压

download

1
tar -zxvf zookeeper-3.4.14.tar.gz

修改配置文件

1
2
3
4
5
6
7
8
cp zoo_sample.cfg zoo.cfg
# 修改存储快照文件snapshot的目录(自己随便修改)
dataDir=/home/hadoop/zookeeper-3.4.14/tmp

# 添加集群节点信息,本机器上必须是0.0.0.0,否则其他机器访问,zk集群起不来,这里server.x中的x是一个数字，与myid文件中的id是一致的
server.1=0.0.0.0:2888:3888
server.2=server02:2888:3888
server.3=server03:2888:3888

创建myid文件

myid文件必须在dataDir目录下

1
2
mkdir /home/hadoop/zookeeper-3.4.14/tmp
echo 1 > /home/hadoop/apps/zookeeper-3.4.9/tmp/myid

启动

1
2
3
4
5
6
7
cd /home/hadoop/zookeeper-3.4.14
# 启动
bin/zkServer.sh start
# 查看状态
bin/zkServer.sh status
# 客户端连接
bin/zkCli.sh

问题:

  • 启动用户
    网上会有说必须用root用户启动,其实并不用
  • Error contacting service. It is probably not running.
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    # zkServer.sh start
    JMX enabled by default
    Using config: /usr/local/zk/bin/../conf/zoo.cfg
    Starting zookeeper … STARTED(正常启动就是这样的显示)

    查看状态却报错：
    # zkServer.sh status
    JMX enabled by default
    Using config: /usr/local/zk/bin/../conf/zoo.cfg
    Error contacting service. It is probably not running.(就是集群没有起来,可以根据zookeeper.out来查看失败原因)

Zookeep启动正常，却报错：Error contacting service. It is probably not running

  • 遇到问题排查思路
    # 确认进程是否启动
[hadoop@server02 zookeeper-3.4.14]$ jps
3216 QuorumPeerMain
3303 Jps
[hadoop@server02 zookeeper-3.4.14]$ netstat -tunlp|grep 3216
# 确认端口是否开放
(Not all processes could be identified, non-owned process info
will not be shown, you would have to be root to see it all.)
tcp        0      0 :::2181                     :::*                        LISTEN      3216/java
tcp        0      0 :::2888                     :::*                        LISTEN      3216/java
tcp        0      0 :::43919                    :::*                        LISTEN      3216/java
tcp        0      0 :::3888                     :::*                        LISTEN      3216/java
# 确认防火墙是否关闭
[hadoop@server01 zookeeper-3.4.14]$ telnet  server02 3888
Trying 100.80.128.165...
Connected to server02.

Zookeeper 集群搭建

发表于 | 分类于
字数统计: | 阅读时长 ≈

分布式文件系统HDFS

HDFS是Hadoop的数据储存部分,主要由NameNode，DataNode,SecondNameNode组成

NameNode

  • NameNode主要功能：接受客户端的读写服务
  • NameNode保存metadata信息包括：
    • 文件ownership和permission
    • 文件包含哪些块
    • Block保存在哪个DataNode（由DataNode启动时上报）
  • NameNode的metadata信息在启动后会加载到内存
    • metadata储存到磁盘文件名为“fsimage”(它是在NameNode启动时对整个文件系统的快照)
    • Block的位置信息不会保存到fsimage
    • edits log(它是在NameNode启动后，对文件系统的改动序列)对metadata的操作日志

SecondNameNode（SNN）

  • SecondNameNode主要功能: 它不是NN的备份（但可以做部分备份），它的主要工作是帮助NN合并edits log，减少NN启动时间
  • SNN执行合并实际
    • 根据配置文件设置的时间间隔fs.checkpoint.period默认3600秒
    • 根据配置文件设置edits log大小fs.checkpoint.size贵的edits文件的最大值默认是64MB

Secondary NameNode:它究竟有什么作用

DataNode（DN)

  • 储存数据（Block）
  • 启动DN线程的时候回向NN回报block信息
  • 通过向NN发送心跳保持与其保持联系（3秒一次），如果NN 10分钟没有收到DN的心跳，则认为其已经lost，并copy其上的block到其它的DN

发表于
字数统计: | 阅读时长 ≈