大数据

stop-all.sh

高容错性

适合批处理

适合大数据处理

可构建在廉价的机器

不适合小文件存取

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

一个文件只能 有一个写者，仅支持 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服务器分开搭建
仍然单独架构存储元数据的数据库

<name>hive.metastore.warehouse.dir</name>
<value>/user/hive/warehouse</value>   

<name>hive.metastore.local</name> 
<value>false</value>   

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聚合会提升 很大的效率

相关的一些参数设置

可能会出现数据倾斜

相关设置

使用场景

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

思路：

缺点;

优化的关键：

分布式协调工具

作用

作用

一般跟datenode存储在同一节点上，可以实现数据的 本地化，便于查询

每个regionserver都会共享一个hlog文件，访问发生错误时可以恢复数据

作用

内存分配

每个region会保存一个表里面某段连续的数据（因为rowkey是按字典序排序的）

region是分布式存储和负载均衡的最小单元，但不是存储的最小单元。region相当于行

每个region是由多个store来组成的，store是由storefile和memstore组成

当一个region越来越大时，就会被split为两个新的region，并有hmaster分配到相应的regionserver

key:  hlogkkey记录数据的归属信息,除了table和region名字外，同时还包括 sequence number和timestamp，timestamp是” 写入时间”

value : HBase的KeyValue对象，即对应HFile中的KeyValue

hlog是一个实现write ahead log的类，每次写入memstore的同时，也会写一份数据到hlog文件中，
hlog文件会定期滚动出新的，并删除旧的文件（已经持久化到storefile的数据）

当hregionserver意外中之后，HMaster会通过Zookeeper感知到，HMaster首先会处理遗留的 HLog文件，
将其中不同Region的Log数据进行拆分，分别放到相应region的目录下，然后再将失效的region重新分配
，领取 到这些region的HRegionServer在Load Region的过程中，会发现有历史HLog需要处理，
因此会Replay HLog中的数据到MemStore中，然后flush到StoreFiles，完成数据恢复

store（保存列族）包括位于一个内存中的memstore和多个位于磁盘的storefile

写数据时，先检查memstore有相同的数据，没有，则先写入hlog，再写入memstore，
当memstore的数据达到阈值（0.9）时，regionserver启动flash ，将数据写入store file。

storefile    当store file文件的数量增长到一定阈值后，会进行合并。合并后的store file文件的大小
和数量超过一定阈值是，会把storefile切分split ，并 把当前的region分为两个，并由
hmaster分配到相应的regionserver，实现符在均衡

集群的名字，是否启用分布式，zookeeper的位置

一般跟datenode保存在相同的节点上，便于数据查询

创建表 

删除表

查看表结构 

查看所有的表 

Put ‘表名’,’rowkey’,’列族:列’,’’列值’

get ‘表名’，‘rowkey’，‘列族：列’

scan ‘表名’   全表扫描 ，慎用

删除某一列

删除一行

删除表中所有数据

删除表中的 数据，并不是直接删除，而是对相应的数据做一个标记，只能在进行storefile合并时进行删除

检索方式

设计rowkey

一般不超过2-3个

当一个store的memstore在做溢写的时候会触发临近列族的meestore也进行溢写，
列族过多时，且有些列族memstore的数量还不是很多的时候，溢写会导致storefile文件零碎过多，
文件合并的效率也会降低。

作用：提供接口关系管理HBase 数据库中的表信息
用法：HBaseAdmin admin = new HBaseAdmin(config);

作用：通过此类可以对HBase进行配置信息
用法实例： Configuration config = HBaseConfiguration.create();

作用：HTable 和 HBase 的表通信
用法：HTable tab = new HTable(config,Bytes.toBytes(tablename));
ResultScanner sc = tab.getScanner(Bytes.toBytes(“familyName”));

作用：HTableDescriptor 包含了hbase中表格的详细信息，包括表中的列族，该表的类型
用法：HTableDescriptor htd =new HTableDescriptor(tablename);
Htd.addFamily(new HColumnDescriptor(“myFamily”));

作用：HColumnDescriptor 维护列族的详细信息，包括列族的版本号，压缩设置等
用法：HTableDescriptor htd =new HTableDescriptor(tablename);
Htd.addFamily(new HColumnDescriptor(“myFamily”));

作用：获取单个行的数据
用法：HTable table = new HTable(config,Bytes.toBytes(tablename));
Get get = new Get(Bytes.toBytes(row));
Result result = table.get(get);

作用：用于向单元格中插入信息
用法：HTable table = new HTable(config,Bytes.toBytes(tablename));
Put put = new Put(row);
p.add(family,qualifier,value);

用于存放 get或者scan操作后的查询结果

/etc/profile

flume-env.sh

可以监听的数据来源

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

可以输出到

telnet 主机 端口 
telnet node02 44444

运行原理

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

比如 ： sqoop list-databases -connect jdbc：mysql：//node01：3306/ -username  -password

sqoop  --options-file  /.. /..   (配置文件的地址) 

在配置文件中指定 导入的目录，还有导出的表的名字，字段

首先将hive 安装在hive所在的客户端上，并进行配置

编写配置文件

配置好导出的文件，导入的表 ，字段

hive建内部表的时候直接将表的映射建立到了hbase中了。因为是hive内部表，
所以建表有hive来建立hbase表，但是如果hive建立外部表，当然就要先有hbase表，
在创建hive的外部表并映射到hbase表。

hive创建表的时候要指定存储方式

指定要与hbase的映射关系

1.hive与hbase配置文件中都需要  jar包   hive-hbase-handle。jar

2.将hbase的相关jar包全部拷贝到hive服务器端的lib'包下

3.在 hive中配置hbase的入口（配置zookeeper）

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-queue
executor处理数据，将结果放入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 分布式远程过程调用

创建的命令

生产者的创建要与broker相连

创建的命令

消费者的创建要与zookeeper相连

每一个消费之在zookeeper中每个唯一保存的元数据信息就是消费者当前消费日志的位移位置。
位移位置是由消费者控制，即、消费者可以通过修改偏移量读取任何位置的数据。

同一个消费组里面的consumer一次不能够去消费同一个Topic的Partition

每个类型的消息成为topic，同一个topic内部的消息按照一定
key和算法备份去存储在不同的broker上

topic 在kafka中可以由多个消费者，订阅消费

创建话题 kafka-topics.sh --zookeeper node02:2181,node03:2181,node04:2181 --create --replication-factor 2 --partitions 3 --topic test
 replication 代表分区的复制个数，默认为一      partition为分区的个数，默认为一    topic 为名字

 每一个分区(partition)都是一个顺序的、不可变的消息队列,并且可以持续的添加,
分区中的消息都被分了一个序列号,称之为偏移量(offset),在每个分区中此偏移量都是唯一的
分区默认保留七天的数据，只需要支持顺序读写文件

分区的方式

分区的组成

分区的目的

读取数据的方法

分区的复制因子

有主构造函数的概念，其他派生的构造函数，就必须直接或者间接的调用主构造函数，
而且调用的时候必须通过关键字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 来开启通信

小案例

spark简介

环境搭建

spark角色分析

spark运行模式

rdd

rdd之间的依赖关系

stage的切割规则

算子

持久化

spark 任务流程