大数据知识点

高容错性

适合批处理

适合大数据处理

可构建在廉价的机器

不适合小文件存取

低延迟数据访问，用大量的时间用于寻址

一个文件只能 有一个写者，仅支持 append

单个namenode的内存压力过大，内存受限

保存文件元数据;单节点

是基于内存存储：不会和磁盘发生交互，但是存储的信息会定时用快照（fsimage）的形式存放在磁盘，做持久化使用

客户端与namenode交互元数据信息

启动过程：

持久化

保存文件的block数据

客户端与namenode交互block数据

第一个副本（这里的源文件就是第一个副本）放置在上传文件的dn，第二个副本放置在与第一个副本不同的机架上，第三个 副本与第二个 副本的机架相同

副本的作用

Client向nameNode发送写数据请求

NameNode节点，记录block信息。并返回可用的DataNode（就近原则）

客户端跟dn交互的时候只个一个dn服务节点交互，然后有这个dn服务与下一个dn服务（备份）建立scoket，在由下一个dn与下下个dn服务建立scoket，这个过程叫pipeline管道模式

client向namenode发送读请求

namenode查看Metadata信息，返回fileA的block的位置

在客户端读取块信息的时候优先选择离自己最近的块信息

作用： 帮助namenode合并edit log 与fsimage（checkpoint操作）

在哪里进行合并日志

快照 fsimage中值保存静态的元数据信息

snn 设置做快照的条件

配置环境变量

先格式化（只在第一次启动过程中做格式化 ）

创建上传文件的目录

上传文件

一 ： namenode启动的时候，会将映像文件（fsimage）载入内存，并执行log文件中的各项操作 完成后生成新的快照，和日志，此实处于安全模式，namenode对于客户端来说是只读的

二： 当上传文件时，副本数小于最小副本数是，认为是安全模式

namenode 单点故障

namenode压力过大，且内存受限，

zookeeper

zkfc（failovercontroller）

journaNades（jn）

namenode

安装过程

配置环境变量

hdfs-site

core-site

启动zookeeper

启动 jn（journalnode）

格式化第一个namenode在、再启动

再同步第二个namenode

初始化zkfc

启动服务

split

map

shuffer（洗牌）

reduce

设置mapreduce基于yarn进行计算

调度的整个过程

resourcemanager

创建application master

设置有关resourcemanager的配置

设置有关zookeeper地址的配置

CLI

Client

Hive Web interface

解释器、编译器、优化器完成HQL查询语句从词法分析、语法分析、编译、优化以及查询计划的生成

会连接 metastore ，hadoop（存储及计算） ，

作用; 客户端连接meta store服务，meta store再去连接mysql数据库来存取元数据。有了meta store服务，就可以有多个客户端同时连接，而且不需要知道mysql的用户名和密码，只要连接metastore即可

metadata（mysql）使用mysql 关系型数据库

存储hive的数据，进行计算的位置

启动metastoreserver hive -- service metastore 9083

启动客户端 hive

启动metastore server hiveserver2 10000（开启thrift服务）

客户端 beeline -u jdbc:hive2://node03:10000/数据库名 -n root 或者 beeline 之后 在输入 jdbc:hive2://node03:10000/数据库名 ；auth=nosasl root mima

将hwi目录进行 打war包

配置hive-site。xml

启动服务

通过网页进入 hive，进行查询

客户端，driver，metastort 布置在用一台机器上

将metastore Server 和hive服务搭建在一起单独架构一台数据库存储元数据

将metastore Server 与hive服务器分开搭建仍然单独架构存储元数据的数据库

hive.metastore.warehouse.dir /user/hive/warehouse

hive.metastore.local false

hive-site。xml

hive --hiveconf hive.cli.print.current.db=true上面的设置是打印当前数据库

hive --hiveconf hive.cli.print.header=true 打印出表的字段名

例如 set hive.cli.print.header=true;

编写文件 .hiverc （宿主 目录下），在启动客户端的时候加载配置通用的hive的环境变量

宿主目录下有一个 隐藏文件 .hivehistory 记录曾经的操作

desc formatted 表名

创建表

删除表

导入本地数据

导入非本地数据

导入数据其实就是文件数据上传到hive'的相关目录下，如果直接将数据拷贝到hive的相关数据目录下，也会被加载为hive数据

插入数据

会将字段的不同值存放在不同的目录下，便于以后的查询效率

静态分区

动态分区

将一个数据文件按照列值取哈希值的方式划分为多个数据文件

可以对每一个表，分区进行分桶，而且还可以根据多了个列进行分桶

可以进行数据抽样检测

增加了join查询的效率

set hive.enforce.bucketing=true;

mr运行时会根据bucket的个数自动分配reduce task的个数一次作业产生的桶（文件数量）和reduce rask的个数一致

CREATE TABLE psnbucket( id INT, name STRING, age INT)CLUSTERED BY (age) INTO 4 BUCKETS ROW FORMAT DELIMITED FIELDS TERMINATED BY ',';

就像动态分区一样，需要中间表导入数据

insert into table bucket_table select columns from tbl;

数据抽样

map-join 关联

例如 ：时间函数，trim函数 ，

自定义的脱敏函数

count，sum，max，min，avg 等函数

一般处理数组，map等

函数 explode （） （一般与split 配合使用，将结果变成多行）

create view as select ......;

drop view 视图名

在指定列上建立索引，会产生一张索引表（Hive的一张物理表），里面的字段包括，索引列的值、该值对应的HDFS文件路径、该值在文件中的偏移量;在执行索引字段查询时候，首先额外生成一个MR job，根据对索引列的过滤条件，从索引表中过滤出索引列的值对应的hdfs文件路径及偏移量，输出到hdfs上的一个文件中，然后根据这些文件中的hdfs路径和偏移量，筛选原始input文件，生成新的split,作为整个job的split,这样就达到不用全表扫描的目的。

每次查询都会先用一个job扫描索引表，如果索引列的值比较稀疏，那么索引表本身就会非常大

索引表不会自动rebuild，如果有数据新增或者删除，必须手动rebuild索引表数据

ALTER INDEX t1_index ON psn2 REBUILD;

索引列的值、该值对应的HDFS文件路径、该值在文件中的偏移量;

show index on 表名

drop index if exists 索引名 on 表名

需要开启metastore和客户端 hive

需要开启metastore和web服务hwi

开启metastore，和hiveserver2，客户端 beeline

在linux下运行 hive 命令

hive -e “sql语句”

hive -e “ sql语句” >文件民 将输出结果重定向到文件

hive -e -S“ sql语句” >文件民 静默输出，将输出结果重定向到文件

hive -f file 运行文件中的sql语句，执行完还在linux端

hive -i file 上同，执行完会进入hive客户端状态

在hive上查看hdfs的内容，也可以运行linux命令

source file （在hive中运行文件）

dfs -ls / （查看hdfs内容）

运行linux 命令 ！pwd

完全兼容SQL的授权模型，推荐使用该模式。就是基于sql授权，grant的授权方式。

权限验证

预防好人做坏事，不是放置坏人做坏事

把hive sql当作mapreduce程序区做优化，最终都是mapreduce任务来执行

explain sql语句； 查看计划

explain extended sql语句； 查看计划的继承

本地模式

集群模式

开启并行计算 ：set hive.exec.parallel=true;

最大允许job个数 ：set hive.exec.parallel.thread.number默认个数为8个，可自行设置

子查询之间没有任何依赖关系，适用于使用并行计算

开启方式

查询限制

目的;

order by

sort by

distribute by

cluster by

join计算时，将小表放在join的左边，减少内存的使用

mapjoin 实现方式

是开启在map端的聚合，省略了suffer和ruduce的过程，在map聚合会提升 很大的效率

相关的一些参数设置

可能会出现数据倾斜

相关设置

使用场景

频繁的申请资源-释放资源，造成资源的浪费，大部分时间都浪费在申请和释放资源过程

思路：

缺点;

优化的关键：

http://prof.ict.ac.cn/publications/

论文：Bigbench:Towards an industry standard benchmark for big data analytics

/etc/profile

flume-env.sh

可以监听的数据来源

管道，用来整理输入和输出的信息

可以输出到

telnet 主机 端口 telnet node02 44444

运行原理

结合 netcat 进行数据的输入，利用telnet协议往netcat里发送数据

有主构造函数的概念，其他派生的构造函数，就必须直接或者间接的调用主构造函数，而且调用的时候必须通过关键字this来操作，而且派生构造函数的名字必须是this。

在类中，除了def定义的成员函数外的所有操作，都可以看作是构造函数的行为组成部分，不管是变量赋值还是函数调用，与java中的class有区别

在scala中，若有继承关系，只有主构造函数才能将参数的值传递到父类的构造函数中去

1.有返回值的函数定义在大括号前必须有等号，不能省略

2.如果函数想要返回值，不要以赋值语句作为最后一条语句

def max（参数）：={函数体 }简写 ： def max（参数）= 函数体

写法 : （）=>{}

匿名函数的调用需要将匿名函数赋值 如 var f= （）=>{}调用时直接调用 f（）

匿名函数 在 高阶函数中 的 写法 （）=>int 返回值类型

先定义一个普通函数，如果说函数的某个变量参数不变，可以给这个函数做包装，让某些参数固定，发生变化的参数继续变化

例子： val fun=showLot(date, _:String) 原来的函数 date是不变的参数 变化的参数用占位符

参数是函数

返回值是函数

参数和返回值都是函数

for循环中写多个条件，相当于嵌套循环

例如：for（i<-1 to 10 ;j<- 1 to 10 ） { } 相当于嵌套循环

a.foreach { x=>{println(x)}}a.foreach{ println(_)}a.freach { println}

var arr=Array(1,2,3,4)

var arr=new Array[Int](3) 3代表数数组的长度 数组元素赋值 arr(0)=2 arr(1)=5

var arr=Array[Array[string]]

var array=Array[Array[String]]( Array[String]("a","b","c"), Array[String]("d","e","f") )

定义 ： var list=List(1,2,3) val list=List("hello neu","hello soft","hello edu","hello dr")

list.map 为每一个输入进行指定的操作，然后为每一条输入返回一个对象，最终输出对象组成的集合

在scala语言中，map可以用来进行键值组队，也可以用来对数据进行批处理在java语言中，用maptopair进行键值组队

list.flatmap 扁平化的map ，对元素进行切分，返回的list 就是单个元素组成的集合 其实是在map的基础上，对所有的对象合并为一个对象

list.filter {x =>x.equals(" ")} 返回的是过滤之后的元素集合

list.count 返回的是 int类型

定义方式 ： var set=Set(1,2,3,4)

两个重要方法

定义方式 ： var map=Map(1->'a',2->'c',(3,'c'),(4,'d') ) 两种写法都一样

遍历map集合元素，相当于得到了 二元组，可以使用（x._1）(x._2) 获取key与value

map 也具有 得到所有的keys或者values

两个map集合相加 ，有相同的key ，先进去的value 留下

var t=new Tuple(1,2,3)

var t=Tuple(1,2,3,4)

var t=(1,2,3,4)

var it=t.productIterator

在Scala中，trait也是有构造代码的，也就是trait中的，不包含在任何方法中的代码// 而继承了trait的类的构造机制如下：1、父类的构造函数执行；2、trait的构造代码执行，多个trait从左到右依次执行；3、构造trait时会先构造父trait，如果多个trait继承同一个父trait，则父trait只会构造一次；4、所有trait构造完毕之后，子类的构造函数执行

写法 :在Scala中，trait是没有接收参数的构造函数的，这是trait与class的唯一区别 所以 一般是 trait Logger { println("Logger's constructor!") }

返回的是true ，因为比较的是内容，不是内存地址。样例类重写了equal，tostring，hashcode 等方法

使用receive 来接受消息（其实还有react ，receiveWithin(5000),reactWithin(5000)）

receive 经常与case 来进行匹配对应的消息

使用 avtor！ 来发送消息

actor通信 类似于线程，通过start 来开启通信

小案例

跨数据源查询引擎

https://github.com/edp963/moonbox

https://github.com/Qihoo360/XSQL

XSQL can be regard as presto implemented by Spark. XSQL is easier to deploy on yarn, more friendly to Spark Programmer

https://github.com/dremio/dremio-oss

子主https://mp.weixin.qq.com/s?__biz=MjM5MDE0Mjc4MA==&mid=2651019031&idx=2&sn=4cd5490a14e1ed5fb5b9bd735f939591&chksm=bdbeaf448ac92652bc994a5b3ec13b5af3470031ff2226184a9e736c0c9262250c660a3f4c6d&scene=27#wechat_redirect题

RAID0

RAID1

RAID5

RAID10

spark简介

环境搭建

spark角色分析

spark运行模式

rdd

rdd之间的依赖关系

stage的切割规则

算子

持久化

spark 任务流程

nimbus（主节点）

zookeeper 分布式资源协调工具

supervisor（从节点）

worker （工作进程）

executor（线程）

task（任务）

spout/bolt

storm采用的是DAG（topology）有向无环图，因此，在storm提交的程序称为topology

storm所处理的最小的单元是tuple（是一个任意对象的数组，如果是自定义的对象 需要进行序列化）

spout理解为数据的来源，bolt是数据的具体处理单元

ZeroMQ开源的消息传递框架，并不是一个MessageQueue

Netty 是基于NIO（同步非阻塞）的网络框架，效率更加高效

storm提交的程序成为topology，--DAG 有向无环图的实现，他处理的最小的消息单元是tuple（任意对象的数组）

从spout中源源不断的传递数据给bolt，以及传递给其他的bolt，形成的数据通道交stream

stream在声明是需要给其指定一个id（默认是default）

生命周期： 拓扑只要启动就会一直在集群中运行 ，直到手动将其kill，否则不会停止。mapreduce执行完就会结束

spout ：数据源 ，一般从外部获取数据， 可以发送多个数据流，需要通过declare方法声明不同的数据流，发送数据时通过SpoutOutputCollector中的emit方法指定数据流Id（streamId）参数将数据发送出去 spout中最核心的方法是nextTuple（）方法，该方法不断的被线程调用，通过emit（）将数据生成Tuple发送出去

bolt：数据流处理组件 ，对接受的数据进行处理， 最核心的方法是execute方法，负责接受一个元组Tuple数据，真正实现核心的业务逻辑 也可以实现多个数据流的发送

zookeeper的目录结构u

nimbus ，supervisor，work 的目录结构

对于每一个worker进程来说，都有一个独立的接收线程和发送线程，负责从网络上发送和接受消息

每个executor线程有自己的incoming-queue和outgoing-queue

worker的接收线程将受到的消息通过task编号传送给对应的executor的incoming-queueexecutor处理数据，将结果放入outgong-queue，达到一定阈值，传输给发送线程

内部通信或在同一个节点的不同worker的thread通信使用LMAX Disruptor来完成。

Disruptor实现了“队列”的功能。可以理解为一种事件监听或者消息处理机制，即在队列当中一边由生产者放入消息数据，另一边消费者并行取出消息数据处理

./bin/storm nimbus >> ./logs/nimbus.out 2>&1 &2>&1 意思是将错误的结果传递给1输出通道，&1表示1 输出通道（标准输出）

提交jar包

a.会把提交的jar包放到nimbus所在服务器的nimbus/inbox目录下 b.submitTopology方法会负责topology的处理；包括检查集群是否有active节点、配置文件是否正确、是否有重复的topology名称、各个bolt/spout名是否使用相同的id等。 c.建立topology的本地目录，nimbus/stormdist/topology-uuid 该目录包括三个文件： stormjar.jar --从nimbus/inbox目录拷贝 stormcode.ser --此topology对象的序列化 stormconf.ser --此topology的配置文件序列化 d.nimbus任务分配，根据topology中的定义，给spout/bolt设置task的数目，并分配对应的task-id,最后把分配好的信息写入到zookeeper的../task目录。 e.nimbus在zookeeper上创建taskbeats目录，要求每个task定时向nimbus汇报 f.将分配好的任务写入到zookeeper,此时任务提交完毕。zk上的目录为assignments/topology-uuid g.将topology信息写入到zookeeper/storms目录

Nimbus和Supervisor甚至实际运行的Worker都是把心跳保存在Zookeeper上的。Nimbus也是根据Zookeerper上的心跳和任务运行状况，进行调度和任务分配的。两者之间的调度器。nimbus和supervisor的所有状态信息都会存放在zookeeper中来管理

a.定期扫描zookeeper上的storms目录，看看是否有新的任务，有就下载。 b.删除本地不需要的topology c.根据nimbus指定的任务信息启动worker

a.查看需要执行的任务，根据任务id分辨出spout/bolt任务 b.计算出所代表的spout/bolt会给哪些task发送信息 c.执行spout任务或者blot任务

LocalCluster cluster=new LocalCluster(); cluster.submitTopology("wc", new Config(), builder.createTopology());

//集群提交 StormSubmitter.submitTopology(args[0], config, builder.createTopology()); //args[0]是传递的参数，别名

builder.setBolt("bolt", new MyBolt(), 2).shuffleGrouping("spout");

builder.setBolt("bolt", new MyBolt(), 2).fieldsGrouping("spout", new Fields("session_id"))

builder.setBolt("bolt", new MyBolt(), 2).globalGrouping("spout");

builder.setBolt("bolt", new MyBolt(), 2).noneGrouping("spout");

一个拓扑会包含一个或者多个worker，每个worker进程只属于一个 topology这些worker并行运行在不同的服务器上

进程数的设置

一个worker中有多个线程，一个线程 中可以执行一个或者多个task任务(但是默认一个线程运行一个task)，每个task任务对应着同一个组件（spout，bolt）

线程数的设置

task是执行数据处理的最小单元，每个task即为一个spout或者一个bolt

task数量在整个topology生命周期中保持不变，executor的数量可以调整

task数量的设置

动态设置拓扑的worker进程数量，以及executor线程数量

设置的代码 ： [storm rebalance topology-name [-w wait-time-secs] [-n new-num-workers] [-e component=parallelism]*]

nimbus与supervisor都是无状态信息和快速启动的，他们的状态信息都保存在zookeeper上

worker挂掉时

nimbus挂掉

nimbus属于单点故障吗

supervisor挂掉

ack机制 ，不能继承baserichspout，要实现irichspout，还要重写 ack与fail方法

配置 storm 的 tuple 的超时时间 – 超过这个时间的 tuple 被认为处理失败了。这个设置的默认设置是 30 秒

ack机制中

ack原理

重复消费

客户端提交数据进行结算，并不会等待数据计算结果

消息是逐条处理

客户端提交数据请求后，立刻取得计算结果并返回客户端

同步处理可以看出是有多个bolt单元并行计算，并行计算性能更好，效率更高

使用到了 DRPC 分布式远程过程调用