大数据基础

什么促进了大数据时代的诞生？

• 第三次信息化浪潮

• 信息科技为大数据时代提供技术支撑，具体是存储容量增大，CPU处理速度提高，网络带宽的提高

• 数据生产方式的变革，运营式系统阶段，用户原则内容阶段，感知式系统阶段

大数据的发展进程

20世纪90年代到21世纪初的萌芽阶段，21世纪前10年的成熟期，2010年之后的大规模应用期

大数据的 4V 概念

4V 就是数据量大Volume，数据类型繁多Variety，处理速度快Velocity，价值密度低Value
数据真实性Veracity

大数据的计算模式

• 批处理计算

• 流计算

• 图计算

• 查询分析计算

云计算

云计算的概念，以虚拟化技术为核心，低成本为目标的动态可扩展的网络基础应用，是近几年来最具代表的网络计算技术和模式

云计算包含了三种服务模式

• IaaS，基础设施即服务

• Paas，平台设施即服务

• SaaS，软件即服务，把软件作为服务进行出租

云计算的关键技术，虚拟化，分布式存储，分布式计算，多租户等

Hadoop

Hadoop是一个由Apache基金会所开发的分布式系统基础架构。用户可以在不了解分布式底层细节的情况下，开发分布式程序。充分利用集群的威力进行高速运算和存储。

Hadoop的框架最核心的设计就是：HDFS和MapReduce

核心组件，HDFS，MapReduce和YARN

优点

1.高可靠性。Hadoop按位存储和处理数据

2.高扩展性。Hadoop是在可用的计算机集簇间分配数据并完成计算任务的，这些集簇可以方便地扩展到数以千计的节点中

3.高效性。Hadoop能够在节点之间动态地移动数据，并保证各个节点的动态平衡，因此处理速度非常快

4.高容错性。Hadoop能够自动保存数据的多个副本，并且能够自动将失败的任务重新分配

5.低成本。hadoop是开源的，项目的软件成本因此会大大降低

6.运行在Linux系统上

7.支持多种编程语言

Hadoop的生态系统

HDFS

Hadoop Distributed File System 缩写 HDFS，它存储 Hadoop 集群中所有存储节点上的文件。基于Google的GFS文件系统开发的，作为Hadoop的底层数据存储

HBase

HBase是一个高可靠的，高性能，可伸缩，实时读写，分布式的列式数据库

MapReduce

Hadoop MapReduce 是针对谷歌的 MapReduce 实现，是一种编程模型，用于大规模数据集的复杂的并行运算。使用到了两个函数 Map() 函数 Reduce() 函数，Map 映射函数，用来把一组键值对映射成一组新的键值对，指定并发的 Reduce 归约函数，用来保证所有映射的键值对中的每一个共享相同的键组。

Hive

hive是基于Hadoop的一个数据仓库工具，用来进行数据提取、转化、加载，这是一种可以存储、查询和分析存储在Hadoop中的大规模数据的机制。

hive数据仓库工具能将结构化的数据文件映射为一张数据库表，并提供SQL查询功能，能将SQL语句转变成MapReduce任务来执行。学习成本低，适合对数据仓库进行统计分析。

zookeeper

ZooKeeper是一个分布式的，开放源码的分布式应用程序协调服务，是Google的Chubby一个开源的实现，是Hadoop和HBase的重要组件。它是一个为分布式应用提供一致性服务的软件，提供的功能包括：配置维护、域名服务、分布式同步、组服务等。

ZooKeeper的目标就是封装好复杂易出错的关键服务，将简单易用的接口和性能高效、功能稳定的系统提供给用户。

ZooKeeper包含一个简单的原语集，提供Java和C的接口。

Hadoop 安装和使用

• 创建Hadoop 用户

• 安装java的环境

• 安装并且设置SSH登录权限

• 单机安装配置

• 伪分布式安装配置

大数据存储和管理

分布式文件系统DFS

目前应用广泛的分布式文件系统主要有GFS和HDFS，分布式文件系统在物理结构上是由计算机集群中的多个节点构成的，这些节点主要分成两类，一类是主节点Master Node 也叫作NameNode名称节点，一类是由Slave Node 也叫作DataNode 数据节点。

NameNode 主要负责文件和目录的创建、删除和重命名，同时管理者数据节点和文件块的映射关系，因此Client只需要访问NameNode节点才可以找到请求的文件块所在的位置，进而从位置读取所需的文件块。

DataNode 主要负责的是数据的存储和读取，在存储时候，由NameNode分配存储位置，Client将数据写入到相应的数据节点。在读取数据时候，Client从NameNode获取到DataNode和文件块的映射关系，就可以根据相对应的位置访问到对应的文件块。

HDFS 是Hadoop的分布式文件系统，实现的目标是，兼容廉价的硬件设备，流数据的读写，大数据集，简单的文件模型，强大的跨平台兼容性。

缺点，不适合低延迟数据访问，无法高效的存储大量小文件，不支持多用户的写入和任意文件的修改。

HDFS 数据块

文件被切分成为固定大小的数据块，默认数据块大小是128MB，Hadoop2.7.3之间是64MB，可配置

数据块大是为了数据传输超过寻道时间，高吞吐量

一个文件按大小切分成为若干个Block，存储到不同的节点上，默认情况下每个块具有三个副本。

Block副本放置的策略

副本1，同Client的节点上

副本2，不同机架的节点上

副本3，与第二个副本同一机架的另一个节点上

计算集群中元数据需要的内存

10600000 * 250Bytes + 310000 * 290Bytes +13300000 * 368Bytes = 7.11GB

• 可扩展性，支持热拔插，可动态的添加和删除DataNode，都会通知NameNode及时更新metedata

• 容错性，多次的数据备份

• 数据局部性，不适合小文件，不能实时的处理实时数据，处理速度慢，缓存效率低下

• 可靠性，冗余副本机制，心跳机制，安全模式，回收站，元数据保护，快照机制

• 高可用性，消除单点故障，防止单个节点宕机之后，造成服务的不可用，保证集群的可用

HBase shell基本命令

https://blog.csdn.net/vbirdbest/article/details/88236575

• 进入Base Shell程序

hbase shell

• help ‘命名名称’ 来查看命令行的具体使用，包括命令的作用和用法。

help 'base'

• 显示集群状态

// 默认 'summary'
status 

• 查看数据库版本

version

• 显示当前用户与组

whoami

• 查看操作表的命令

table_help

• 退出HBase Shell

 exit

• 创建表

create '表名', {NAME => '列族名1'}, {NAME => '列族名2'}, {NAME => '列族名3'}
// 此种方式是上面的简写方式，使用上面方式可以为列族指定更多的属性，如VERSIONS、TTL、BLOCKCACHE、CONFIGURATION等属性
// 推荐
create '表名', '列族名1', '列族名2', '列族名3'

create 'tbl_user', 'info', 'detail'

• 列出所有的表

list

• 修改表的结构

alter '表名', '列族名'

alter 'tbl_user', 'address'

• 删除一个列族

alter '表名', {NAME=> '列族名', METHOD=> 'delete'}

alter 'tbl_user', {NAME=> 'address', METHOD=> 'delete'}

• 获取表的描述

describe 'table_name'

• 表是否存在

exists '表名'

exists 'tbl_user'

• 启用表和禁用表

enabled 'table_name'

disabled 'table_name'

查看表是否被禁用

is_enabled 'table_name'

is_disabled 'table_name'

• 删除表

先禁用，再删除

disable 'tbl_user'
drop 'tbl_user'

• 插入和修改数据

# 第一行数据
put 'tbl_user', 'mengday', 'info:id', '1'
put 'tbl_user', 'mengday', 'info:name', '张三'
put 'tbl_user', 'mengday', 'info:age', '28'

put 'tbl_user', 'mengday', 'detail:birthday', '1990-06-26'
put 'tbl_user', 'mengday', 'detail:email', 'abc@163.com'
put 'tbl_user', 'mengday', 'detail:create_time', '2019-03-04 14:26:10'

put 'tbl_user', 'mengday', 'address', '上海市'

# 第二行数据
put 'tbl_user', 'vbirdbest', 'info:id', '2'
put 'tbl_user', 'vbirdbest', 'info:name', '李四'
put 'tbl_user', 'vbirdbest', 'info:age', '27'

put 'tbl_user', 'vbirdbest', 'detail:birthday', '1990-06-27'
put 'tbl_user', 'vbirdbest', 'detail:email', 'xxx@gmail.com'
put 'tbl_user', 'vbirdbest', 'detail:create_time', '2019-03-05 14:26:10'

put 'tbl_user', 'vbirdbest', 'address', '北京市'

# 第一行数据
put 'tbl_user', 'xiaoming', 'info:id', '3'
put 'tbl_user', 'xiaoming', 'info:name', '王五'
put 'tbl_user', 'xiaoming', 'info:age', '26'

put 'tbl_user', 'xiaoming', 'detail:birthday', '1990-06-28'
put 'tbl_user', 'xiaoming', 'detail:email', 'xyz@qq.com'
put 'tbl_user', 'xiaoming', 'detail:create_time', '2019-03-06 14:26:10'

put 'tbl_user', 'xiaoming', 'address', '杭州市'

• 全表扫描

scan 'table_name'

• 扫描表的列簇

scan 'table_name', {COLUMN=>'info'}

• 扫描列簇的某个列

scan 'tbl_user', {COLUMN=>'info:age'}

• 获取簇数据和获取簇中的列数据

get '表名', '行键'

get '表名', '行键', '列族名'

• 删除列族的某个列

delete '表名', '行键', '列族名:列名'

• 删除某行数据

deleteall '表名' '行键'

• 清空表的数据

truncate 'table_name' 

• 查询表有多少行

count 'table_name'

• 时间范围

# 语法
scan '表名',{TIMERANGE=>[timestamp1, timestamp2]}

# 示例
scan 'tbl_user',{TIMERANGE=>[1551938004321, 1551938036450]}

NoSQL数据库

NoSQL数据库的简介和兴起的原因

NoSQL，泛指非关系型的数据库。NoSQL数据库的产生就是为了解决大规模数据集合多重数据种类带来的挑战，特别是大数据应用难题。

特点，灵活可扩展，灵活的数据模型，与云计算紧密相连

兴起的原因，海量数据的需求，数据高并发的需求，无法满足可扩展和高可用性的需求

非关系型数据库和关系型数据库的区别

NoSQL 的四大数据类型

• 键值对数据类型

哈希表，这个表中有一个特定的键和一个指针指向特定的数据。
比如，Tokyo Cabinet/Tyrant， Redis， Voldemort， Oracle BDB。

• 列存储数据库

通常是用来应对分布式存储的海量数据。键仍然存在，但是它们的特点是指向了多个列。这些列是由列家族来安排的。如：Cassandra， HBase， Riak.

• 文档型数据库

灵感是来自于Lotus Notes办公软件的，而且它同第一种键值存储相类似。该类型的数据模型是版本化的文档，半结构化的文档以特定的格式存储，比如JSON。在处理网页等复杂数据时，文档型数据库比传统键值数据库的查询效率更高。如：CouchDB， MongoDb. 国内也有文档型数据库SequoiaDB，已经开源。

• 图形型数据库

是使用灵活的图形模型，并且能够扩展到多个服务器上

NoSQL 三大基石

三大基石是，CAP，BASE理论，最终一致性

• CAP

C，Consistency 一致性，分布式系统的所有数据备份中，在同一时刻是否相同的值

A，Availability 可用性，保证每一个请求不管成功还是失败都具有响应

P，Partition toleration 分区容错性，系统中任意信息的对视或者失败都不会影响系统的继续运行

CAP原则就是这三个元素最多只能实现两个，不可能三者兼顾

• BASE理论

BASE是Basically Available（基本可用）、Soft state（软状态）和Eventually consistent（最终一致性）三个短语的简写

基本可用（Basically Available）
指一个分布式系统的一部分发生问题变得不可用时，其他部分仍然可以正常使用，也就是允许分区失败的情况出现。

例如：一个数据库系统部署了很多节点，有可能一两个节点出现了失败，但是整个系统依然是可用的

软状态（Soft State）：
与硬状态是对应的，指状态可以有一段时间不同步，具有不同步的时间窗口，具有一定滞后性。

补充：硬状态（Hard State）:数据库状态必须一直保持数据库状态一致性，就是指任意时刻数据必须时正确的。

最终一致性（Eventual Consistency）
一致性的类型包括“强一致性”和“弱一致性”，两者的主要区别在于高并发的数据访问操作下，后续操作是否能够获取最新的数据。最终一致性是为弱一致性的一种特例。

强一致性：一般要求，执行完一个更新之后，后续的其他操作都能够读到你更新的数据
弱一致性：反之，执行一个更新操作之后，后续的读操作不能保证马上读到你刚写入的数据

• 最终一致性

最终一致性为弱一致性的一种特例。根据后台各个进程访问到数据的时间和方式不同，可以区分为

因果一致性

进程A更新了数据，通知B进程，B进程就可以访问到更新的值，没有通知，最终可以访问到，不是马上就访问到

读自己之所写

进程A更新的值后，A进程访问数据是更新过后的值

单调读写一致性

进程A看到过数据对象的某个值，那么后续访问都不会返回在那个之前的旧值

会话一致性

读自己所写的一致性

单调写一致性

保证来自同一进程的写操作按照顺序执行

World Count的MapReduce处理

统计单词个数

    public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable>{

        private final static IntWritable one = new IntWritable(1);
        private Text word = new Text();

        public void map(Object 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);
            }
        }
    }

    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);
        }
    }

统计单词的长度数量，map函数和reduce函数

    public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {
        private final static IntWritable one = new IntWritable(1);
        private Text category = new Text();

        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            StringTokenizer itr = new StringTokenizer(value.toString()," ,:?;\\.\t\n\r\f" ); //
            while (itr.hasMoreTokens()) {
                String w = itr.nextToken();
                int length = w.length();
                String c;
                //w="come" -->"small"
                if(length == 1){				//w.startsWith("C") w.length() == 4
                    c ="tiny";
                }
                else if(length >= 2 && length <=4){				//w.startsWith("C") w.length() == 4
                    c ="small";
                }
                else if(length >= 5 && length <=9){				//w.startsWith("C") w.length() == 4
                    c ="medium";
                }
                else
                {
                    c ="big";
                }
                category.set(c);
                context.write(category, one);
            }
        }
    }
    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);
        }
    }

Flink的容错机制

Exactly once，是指每条 event 会且只会对 state产生一次影响，这里的“一次”并非端到端的严格一次，而是指在Flink内部只处理一次，不包括source和 sink的处理。

At least once，是指每条event 会对state产生最少一次影响，也就是存在重复处理的可能。

At most once，是指每条event 会对state产生最多一次影响，就是状态可能会在出错时丢失。

Flink是如何实现End-To-End Exactly-once的?

Flink通过状态和两次提交协议来保证了端到端的exactly-once语义 Source支持数据的replay，如Kafka的offset。

Transformation:借助于checkpoint

Sink: Checkpoint+两阶段事务提交

Spark 计算 Pi

在idea 中编写spark计算Pi值

object ScalaPi {
  def main(args: Array[String]) {
    //创建一个Config
    val conf =
      new SparkConf()
        .setAppName("ScalaPI")
        .setMaster("local")

    //核心创建SparkContext对象
    val sc = new SparkContext(conf)

    //计算PI
    val count = sc.parallelize(1 to 10000).filter { _ =>
      val x = math.random
      val y = math.random
      x * x + y * y < 1
    }.count()
    println(s"Pi is roughly ${4.0 * count / 10000}")

    //停止SparkContext对象
    sc.stop()
  }
}