使用条件:

• 本机的pycharm必须是专业版本
• 远程服务器中需要启动python所依赖的其它服务(mysql,redis…)
• 远程的python服务,由本机来启动

使用设置:

1. 上传代码到服务器上
   • Tool–>Deployment–>Configuration–>+–>Type:SFTP(相当于ssh)–>输入配置,测试是否连接–>Mappings(本地目录与远程目录的映射)
2. 调试(设置远程解释器)
   • File–>Setting–>Project–>Project Interpreter(设置成远程服务器的python解释器)–>add remote–>SSh Credentials–>输入配置
3. 配置远程服务器运行ip和端口
   • Run–>Edit Configuration–>Django server–>Host(0.0.0.0:80)–>Project Interpreter(选择远程的解释器)

遇到的坑:

• 2018.04的专业版本不行之后,只能换成2017的版本

./pycharm.sh安装之后,uninstall
find ~/ -name PyCharm*

nginx 基本知识

location 匹配url

• 优先级:
  • [精确匹配(同等类型顺序越前,优先级越高)]
  • [开头匹配(同等类型长度越长,优先级越高)]
  • [正则匹配(同等类型顺序越前,优先级越高)]
  • [通用匹配(同等类型长度越长,优先级越高)]
• 语法规则：location [=|^~|~|~*] /uri/ {…} 先匹配普通location，在匹配正则location

设置静态资源的过期时间

• 服务器静态资源是否过期 存在两个判断机制的：

  • 新鲜度检测阶段:需要依赖响应头的Cache-Control/Expires，通过才会返回200(from cache)
  • 资源二次校验阶段:校验资源一致性,Last-Modified/ETag 这两个数据用于确定数据是否改变.如果一致返回304,如果不一致返回200,当然，如果没有找到返回404

• expires指令用来对浏览器本地缓存的控制,指令可以放置在http {}，server {}，location {}或位置{}块内的if语句中

  • 当没有设置expires指令时:

curl -I http://0.0.0.0/logo.png
HTTP/1.1 200 OK
Server: nginx/1.10.3 (Ubuntu)
Date: Fri, 22 Jun 2018 03:16:20 GMT
Content-Type: image/png
Content-Length: 18
Connection: keep-alive
Pragma: public
Cache-Control: public

• • 当设置expires指令时:

curl -I http://0.0.0.0/logo.png
HTTP/1.1 200 OK
Server: nginx/1.10.3 (Ubuntu)
Date: Fri, 22 Jun 2018 03:16:20 GMT
Content-Type: image/png
Content-Length: 18
Connection: keep-alive
Pragma: public
Cache-Control: public
Expires: Sun, 22 Jul 2018 03:16:20 GMT
Cache-Control: max-age=2592000

• 遇到的坑(一直没有效果)

  • 想着是否还需要额外的模块支持
  • 参数是否增加的不够全(还要加请求头)
  • location匹配是否有错误
  • 各种变着法的换其它各种形式的写法

• server_name问题(要写明name,不然只会匹配 location / {…}规则)

  • 没有添加 0.0.0.0
    • http://0.0.0.0/ 可正常访问
    • http://localhost/ 可正常访问
    • http://127.0.0.1/ 可正常访问
    • http://0.0.0.0/hello 不可正常访问
    • http://localhost/hello 可正常访问
    • http://127.0.0.1/hello 不可正常访问
  • 新添加 0.0.0.0后
    • http://0.0.0.0/ 可正常访问
    • http://localhost/ 可正常访问
    • http://127.0.0.1/ 可正常访问
    • http://0.0.0.0/hello 可正常访问
    • http://localhost/hello 可正常访问
    • http://127.0.0.1/hello 不可正常访问

• 解决的方法:原因居然是server_name 没有加0.0.0.0(在命令行和浏览器中测试都是使用0.0.0.0)

缓存服务器设置(配置proxy_cahe):只是针对API后端,所以才需要 proxy_pass或者 uwsgi_pass配置才生效

• 参数说明:

• 遇到的坑(也一直没有效果)

  • proxy_pass http://0.0.0.0/; # uwsgi启动的话,则将对应的proxy设置换成uwsgi即可

• 查看是否生效

  curl -I http://0.0.0.0/logo.png
  X-Cache: MISS
  X-Cache: HIT  # 第二次请求,如果是HIT,则表示缓存生效
  

最终配置如下:

proxy_cache_path /tmp/nginx levels=1:2 keys_zone=cache:10m inactive=60m;

map $sent_http_content_type $expires {
    default                    off;
    text/html                  epoch;
    text/css                   max;
    application/javascript     max;
    image/png                  2d;
    ~*\.(jpg|jpeg|png|gif|ico|css|js|pdf)(\?|$) 7d;

}

server {
    listen       80;
    server_name 0.0.0.0 localhost ;
    location = /hello {
        return 302  https://www.baidu.com;    
    }

    # expires 1d;

    # location ~* .*\.(ico|css|js|gif|jpe?g|png|woff|woff2|eot|svg|ttf|)$ {
        # expires 1d;
    # }

    # location ~* \.(jpg|jpeg|png|gif){
        # expires 1d;
    # }

    # location = /logo.png {
        # root /usr/share/nginx/html;
        # expires 30d;
        # access_log off;
        # log_not_found off;
        # add_header Pragma public;
        # add_header Cache-Control "public"; # 任何系统都可以缓存它们
    # }

    # expires $expires; # 这种方式在chrome,firefox没有Expires,在命令行中就有
    location / {

        root /usr/share/nginx/html;
        index index.html;
        if ($request_uri ~* "\.(ico|css|js|gif|jpe?g|png)$") {
            expires 30d;
            access_log off;
            add_header Cache-Control "public";
        }


        proxy_cache cache;
        proxy_cache_valid 200 301 304 30m;
        proxy_pass    http://0.0.0.0:8000;

        # 缓存 key 的值
        proxy_cache_key $host$uri$is_args$args;
        # proxy_cache_key "$scheme$request_method$host$request_uri";
        proxy_cache_use_stale updating error timeout invalid_header http_500 http_502 http_503 http_504;

        expires 1m;
        # 将缓存的状态添加到 Header 中
        add_header X-Cache '$upstream_cache_status';
    }
}

缓存服务器设置发生在:API后端和Nginx静态文件分离情况的配置文件(适用后端接口不是实时的改变的)

uwsgi_cache_path /tmp/nginx_cache/ levels=1:2 keys_zone=mycache:10m max_size=10g inactive=10m use_temp_path=off;

# the upstream component nginx needs to connect to
upstream uwsgi {
    server unix:/code/app.sock; # for a file socket
}

# configuration of the server
server {
    # the port your site will be served on
    listen    80;
    server_name  0.0.0.0;
    charset     utf-8;
    # add_header Accept-Ranges bytes;
    # add_header Content-Encoding identity;
    # add_header Accept-Encoding identity;
    # add_header Range bytes=0-;

    client_max_body_size 75M;   # adjust to taste


    # Django media
    location /media  {
        alias /code/media;  # your Django project's media files - amend as required
        if ($request_uri ~* "\.(ico|gif|jpe?g|png|woff|woff2|eot|svg|ttf|mp4)$") {
            expires 3d;
            access_log off;
            add_header Cache-Control "public";

        }

    }

    location /static {
        alias /code/static; # your Django project's static files - amend as required

        if ($request_uri ~* "\.(css|js)$") {
            expires 1d;
            access_log off;
            add_header Cache-Control "public";
        }
    }

    location /favicon.ico {
        alias /code/media/logo.png; # your Django project's static files - amend as required
    }

    # / 必须放在最后,不能覆盖静态url配置
    location / {

        add_header      X-uWSGI-Cache $upstream_cache_status;

        # server cache
        uwsgi_cache  mycache;
        uwsgi_cache_valid  200 301 1m;
        slice 1m;
        proxy_set_header Range $slice_range;
        proxy_set_header Range $http_range;
        # proxy_cache_valid 200 301 304 30m;

        uwsgi_cache_key  $request_uri;

        # client cache
        expires 3d;

        uwsgi_pass  uwsgi;
        include     /etc/nginx/uwsgi_params; # the uwsgi_params file you installed
    }
}

查看服务器的基本信息

# 总核数 = 物理CPU个数 X 每颗物理CPU的核数 
# 总逻辑CPU数 = 物理CPU个数 X 每颗物理CPU的核数 X 超线程数
# 一个物理封装的CPU(通过physical id区分判断)可以有多个核(通过core id区分判断).而每个核可以有多个逻辑cpu(通过processor区分判断)

# 查看物理CPU个数(主机上看得见摸得着那块CPU硬件)
cat /proc/cpuinfo| grep "physical id"| sort| uniq| wc -l

# 查看每个物理CPU中core的个数(即核数:处理数据的芯片组数量,双核,四核就是指的CPU核心)
cat /proc/cpuinfo| grep "cpu cores"| uniq

# 超线程:一个CPU核就是一个物理线程,由英特尔开发超线程技术可以把一个物理线程模拟出两个线程来使用,使得单个核心用起来像两个核一样,以充分发挥CPU的性能

# 查看逻辑CPU的个数(top命令,然后输入1也是可以的)
cat /proc/cpuinfo| grep "processor"| wc -l

# 查看CPU信息(逻辑核数和型号)
cat /proc/cpuinfo | grep name | cut -f2 -d: | uniq -c

# 系统简单查看CPU
lscpu

# 查看内存 cat /proc/meminfo
free -m

# 查看磁盘空间
df -h 

# 查看目录大小
du -sh ./ # (目录名)

# 查看用户登录日志
last

# 查看网速
pip install speedtest-cli
speedtest-cli # 上行/下行网速

#磁盘的读取速度
yum install hdparm
hdparm -Tt /dev/vda1

进程管理工具之ps

ps # 仅仅显示本终端的进程

-a # 显示终端中进行的所有进程
-x # 会显示没有控制终端的进程
-e # 所有的进程
-u # 查看某个用户的所有进程
-f # 来查看格式化的信息列表
ps -u femn # 查看所有femn用户的进程 
ps aux # 可以对系统进程更加全面的了解
ps -aux | less 
# less是一个分页显示文件的工具工具,它允许你一页一页(或一个屏幕一个屏幕)地查看信息

ps -ef # 所有进程 PID PPID 
ps -efH # 把输出的进程组成一个层级的格式 树状 
ps axjf  # 树状视图 等于 pstree
ps -ef | grep python3 | cut -c 10-15 | xargs kill -9 
# Kill 某个用户或命令的所有进程
ps axjf |grep nginx
pgrep nginx # 单单的得到PID,和其子进程号

ps -aux --sort -pcpu | head -n 10
# 根据 CPU 使用来升序排序 +pcpu倒序
ps -aux --sort -pmem | head -n 10
# 根据 内存 使用来升序排序
ps -aux --sort -pcpu,-pmem | head -n 10

# 仅仅得到本机的程序名相关的进程
ps aux/-ef |grep ssserver # 程序名(命令名)或者PID

进程管理工具之top

top操作:

前五行是系统整体的统计信息.第一行是任务队列信息,同 uptime 命令的执行结果

第二三行为进程和CPU的信息.当有多个CPU时,这些内容可能会超过两行.内容如下:

最后两行为内存信息:

top中显示状态的字段

ab

由于对于并发请求,cpu实际上并不是同时处理的,而是按照每个请求获得的时间片逐个轮转处理的,所以基本上第一个Time per request时间约等于第二个Time per request时间乘以并发请求数

并发继续增大后nginx遇到的第一个瓶颈在CPU

网络相关的服务(socket服务),或其它的网络相接进程(网络是双向的)

netstat -anpt |grep ssh #程序名  得到PID 端口号
netstat -lnpt |grep 8388  #端口号 得到PID,进程名
sudo lsof -i :8001   #端口号 PID,User,进程名
sudo lsof -p 1609    #PID 

ll /proc/进程号  
killall http* 它支持通过进程名而不是进程号来结束进程,也支持通配符

storm:分布式实时分析计算系统

storm概念

Topologies:拓扑,也称为一个任务
Spouts:集群i(拓扑)的消息源
Bolts:集群(拓扑)节点的处理逻辑单元
Configuration:topology配置

tuple:消息元组(在Spouts和Bolts之间传递的数据格式,一种自定义格式的封装)
Stream:流,tuple(消息的处理)经过的路径不一样
Stream groupings:流的分组策略

Tasks:任务处理单元
Executor:工作线程
Workers:工作进程

搭建storm集群

先在cluster1-3中安装zookeeper集群

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

# 怕有冲突 先将zookeeper目录,在cluster1-3中执行
tar -zcvf z.tar.gz zookeeper-3.4.10
rm -rf /root/zookeeper-3.4.10
# 在cluster1中执行
tar -zxvf zookeeper-3.4.10.tar.gz
cd zookeeper-3.4.10/conf/
mv zoo_sample.cfg zoo.cfg
vim zoo.cfg 
    dataDir=/root/zookeeper-3.4.10/data
    server.1=cluster1:2888:3888
    server.2=cluster2:2888:3888
    server.3=cluster3:2888:3888
cd ../ && mkdir data && cd data 
echo 1 > myid
cd 
# 将ZK复制到cluster2和3
scp -r ./zookeeper-3.4.10 root@cluster2:/root
scp -r ./zookeeper-3.4.10 root@cluster3:/root
# 分别在cluster2 
echo 2 > /root/zookeeper-3.4.10/data/myid
# 分别在cluster3 
echo 3 > /root/zookeeper-3.4.10/data/myid
# 分别在cluster1-3运行
zkServer.sh start

复制安装包

1

scp ./apache-storm-1.1.1.tar.gz root@cluster1:/root/

cluster1在运行

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

tar -zxvf apache-storm-1.1.1.tar.gz
mv apache-storm-1.1.1 storm
cd storm/conf 
vim storm.yaml
    storm.zookeeper.servers:
        - "cluster1"
        - "cluster2"
        - "cluster3"
    nimbus.seeds: ["cluster1"]
    storm.zookeeper.root: "/root/zookeeper-3.4.10"
cd 
scp -r storm root@cluster2:/root/
scp -r storm root@cluster3:/root/
# 配置环境变量(cluster1-3)
    echo "export PATH=\${PATH}:/root/storm/bin" >> ~/.bashrc
    source ~/.bashrc

在nimbus主机上运行(cluster1)

1
2

nohup storm nimbus 1>/dev/null 2>&1 &
nohup storm ui 1>/dev/null 2>&1 &

访问 http://cluster1:8080/

在supervisor主机上运行(cluster2-3),等nimbus跑起来之后 加一个就让HMaster管理

1

nohup storm supervisor 1>/dev/null 2>&1 &

hadoop工具(二):hbase也是分布式,可扩展的大数据存储,HBase与HDFS的区别在于它是个数据库,数据库是基于文件系统(HDFS)提供的一种结构化数据的查询管理的一个系统
应用场景:
对数据进行随机的,实时的进行读写
相对于处理复杂的表与表之间的关系,以及关联查询的关系型数据库而言.HBASE是没有表与表的关联,是属于NoSql的范围

HBASE表结构:

1. 建表时,不需要限定表中的字段,只需要指定若干个列族
2. 插入数据时,一条数据中可以存储任意多个列(K-V,列名&列值),不在乎是否冗余,只要查询方便
3. 一个Value可以有多个版本,通过版本号来区分(时间戳),要查询某一个具体字段的值,需要指定的坐标:表名–行键–列族(ColumnFamily)–列名(Qualifier)(sql中的字段)–版本
4. 行键是唯一的

SQL记录的信息,只能是表中已经定义好的字段,而不能插入没有定义的字段,除非改变整个表结构

HBase是一很大很的表(big table),不可能是单机存储,是分布式管理和存储的
只要数据达到一定量之后,若干行就会切分成一个Region,Region会放在Region Server中

安装运行hbase,建立在集群之上

在cluster3中安装hbase
scp ./hbase-1.2.6-bin.tar.gz root@cluster3:/root

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
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

cd /root 
tar -zxvf hbase-1.2.6-bin.tar.gz
cd hbase-1.2.6 
rm -rf docs
cd conf
vim hbase-env.sh
    # 告诉HBase使用外部的ZK
    export HBASE_MANAGES_ZK=false
vim hbase-site.xml
    <configuration>
    <property>
        <name>hbase.rootdir</name>
        <value>hdfs://ns1/hbase</value>
    </property>
    <property>
        <name>hbase.cluster.distributed</name>
        <value>true</value>
    </property>
    <property>
        <name>hbase.zookeeper.quorum</name>
        <value>zookeeper1:2181,zookeeper2:2181,zookeeper3:2181</value>
    </property>
    <property>  
        <name>hbase.master.info.port</name>  
        <value>60010</value>  
    </property> 
    </configuration>
vim regionservers
    zookeeper1
    zookeeper2
    zookeeper3
cp ~/hadoop-2.8.1/etc/hadoop/{core-site.xml,hdfs-site.xml} ./
cd 
mv hbase-1.2.6 hbase
mv hbase hadoop-2.8.1/
scp -r hadoop-2.8.1/hbase zookeeper1:/root/hadoop-2.8.1
scp -r hadoop-2.8.1/hbase zookeeper2:/root/hadoop-2.8.1
scp -r hadoop-2.8.1/hbase zookeeper3:/root/hadoop-2.8.1
scp -r hadoop-2.8.1/hbase cluster4:/root/hadoop-2.8.1
# 已经在zookeeper1-3启动 
zkServer.sh start 
hadoop-daemon.sh start journalnode
# 在cluster1启动
start-dfs.sh
# 在本机(cluster3:HMaster)
cd /root/hadoop-2.8.1/hbase/bin
./start-hbase.sh 
# 在cluster4中执行
cd /root/hadoop-2.8.1/hbase/bin/
./hbase-daemon.sh start master
./hbase shell
version
# 访问http://cluster3:60010/

HDFS元数据的可靠性有保证,但hadoop的HA(高可用)不高

HDFS HA高可用的集群搭建

7台机器的集群规划(512M,8G)

cluster1 192.168.1.221 jdk,hadoop namenode zkfc
cluster2 192.168.1.222 jdk,hadoop namenode zkfc
cluster3 192.168.1.223 jdk,hadoop resourcemanager
cluster4 192.168.1.224 jdk,hadoop resourcemanager
zookeeper1 192.168.1.225 jdk,hadoop,zookeeper zookeeper journalnode datanode nodemanager
zookeeper2 192.168.1.226 jdk,hadoop,zookeeper zookeeper journalnode datanode nodemanager
zookeeper3 192.168.1.227 jdk,hadoop,zookeeper zookeeper journalnode datanode nodemanager

实际操作(之后的所有操作都是root用户)

前提

1. 修改了网卡设置,网络正常.
2. 在/root目录下有hadoop-2.8.1.tar.gz,jdk8.tar.gz和zookeeper-3.4.10.tar.gz这三个包

在一台虚拟机上执行如下脚本(cluster1)

1

curl https://file.femnyy.com/file/install_hadoop_cluster.sh | sudo sh

然后就clone 6台虚拟机,进入开启这6台虚拟机,修改IP,重新启动网络

1
2
3
4

vim /etc/sysconfig/network-scripts/ifcfg-enp0s3
vim /etc/sysconfig/network-scripts/ifcfg-enp0s8
service network restart
ifconfig

然后在cluster1中执行SSH免密码登陆,分别对列出的主机执行

1

ssh-copy-id cluster1 cluster2 zookeeper1 zookeeper2 zookeeper3

在cluster3中执行如下命令,分别对列出的主机执行

1

ssh-copy-id zookeeper1 zookeeper2 zookeeper3

在zookeeper2在执行如下命令

1

echo 2 > /root/zookeeper-3.4.10/data/myid

在zookeeper3在执行如下命令

1

echo 3 > /root/zookeeper-3.4.10/data/myid

环境配置好了之后,一定严格按下面步骤进行操作

启动zookeeper集群(分别在zookeeper1-3上启动zk)

1
2
3

zkServer.sh start
# 查看状态,一个leader(通过选举得到),两个follower
zkServer.sh status

启动journalnode(分别在zookeeper1-3上执行)

1
2
3

hadoop-daemon.sh start journalnode
# 检验是否有JournalNode进程
jps

格式化HDFS(在cluster1上执行)

1
2
3
4
5

hdfs namenode -format
# 格式化后会在根据core-site.xml中的hadoop.tmp.dir配置生成个文件,我是设置成/root/hadoop-2.8.1/tmp,然后将这个tmp目录cp到cluster2下
scp -r /root/hadoop-2.8.1/tmp/ root@cluster2:/root/hadoop-2.8.1/
# 或者在cluster2执行如下命令,效果也是一样的
# hdfs namenode -bootstrapStandby

格式化ZKFC(在cluster1上执行即可)

1
2
3
4
5
6

hdfs zkfc -formatZK
# 在zookeeper1在执行,查看建立的数据节点
zkCli.sh
ls / # [zookeeper, hadoop-ha]
ls /hadoop-ha # [ns1]
get /hadoop-ha/ns1 # 因为没有运行,数据都是空的

如果改了配置文本想重新启动就不需要再启动上面的内容了,
比如改了所有虚拟机的hdfs-site.xml文件,只需要在cluster1 stop-dfs.sh 再运行start-dfs.sh就可以了

如是下次重新启动时就启动zk集群和start-dfs.sh和start-yarn.sh,格式化一次就可以了

启动HDFS(在cluster1上执行即可)

1

start-dfs.sh

启动YARN(在cluster3上执行)

将Namenode和Resourcemanager分开是因为性能问题

1

start-yarn.sh

启动YARN daemon(在cluster4上执行)

1

yarn-daemon.sh start resourcemanager

访问
HDFS: http://cluster1:50070
HDFS: http://cluster2:50070
YARN: http://cluster3:8088

改进: dfs.datanode.http.address datanode的HTTP服务器和端口 50075hdfs-site.xml 0.0.0.0:50075

HDFS HA 测试

测试上传

在zookeeper1-3集群中在执行

1

cd /root/hadoop-2.8.1/tmp/dfs/data/current/BP-xx/current/finalized/

在cluster1在执行

1
2
3
4

hadoop fs -put /root/hadoop-2.8.1.tar.gz /
# 执行完成后会在zookeeper1-3中 cd subdir0/subdir0/看到两个block块
mkdir 1 && cd 1
hadoop fs -get /hadoop-2.8.1.tar.gz

测试namenode切换

HDFS: http://cluster2:50070
HDFS: http://cluster2:50070
得知哪个是active状态,我的是cluster2是active
通过jps得到namenode的进程,将其kill

kill cluster2 namenode 进程

1
2
3

#  kill之后,可以再看http://cluster2:50070的和cluster1的状态,成功切换
# 所以也是在cluster2中重新启动NameNode
hadoop-daemon.sh start namenode

将cluster1这个虚拟机关机

现在是cluster1为active,,cluster2至少要等30s才能切换.
然后重新启动cluster1之后执行

1
2

hadoop-daemon.sh start zkfc
hadoop-daemon.sh start namenode

手动切换,但不推荐使用

现在cluster2是active,在cluster1中执行

1
2
3

hdfs haadmin -transitionToStandby nn2 --forcemanual
# namenode的状态
hdfs haadmin -getServiceState nn2 # standby

查看Yarn状态

在zookeeper1-3的随便一台在执行

1
2
3

zkCli.sh
get /yarn-leader-election/yrc/ActiveBreadCrumb 
# yrcrm1表明rm1(cluster3)是运行状态

也可以通过访问:
YARN:http://cluster3:8088
YARN:http://cluster4:8088
会自动跳转到运行状态的地址上

Yarn的HA测试

我的是cluster3是active,在cluster4在执行

1

hadoop jar hadoop-2.8.1/share/hadoop/mapreduce/hadoop-mapreduce-examples-2.8.1.jar pi 5 5

在cluster3在杀死ResourceManager进程

1
2

#重启启动 resourcemanager
yarn-daemon.sh start resourcemanager

Yarn的HA不像HDFS的YA一样,YARN正在跑的程序突然中断则意味着程序的失败

动态增加DataNode节点和数量管理

在zookeeper3中kill掉datanode进程
再查看http://cluster1:50070的Live Nodes数量,则会显示Live NOdes:2,Dead NOdes:1,block副本不会变多

1
2
3

# 在zookeeper1-3中
vim /root/hadoop-2.8.1/tmp/dfs/data/current/VERSION
# 是由clusterID的决定这个datanode是否在同一个集群上

新加一个datanode节点,只需要hadoop包,然后启动datanode就行了

clone一台zookeeper1的名为zookeeper4,并在zookeeper4运行

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

# zookeeper4 192.168.1.228 jdk,hadoop,zookeeper zookeeper journalnode datanode nodemanager 
# vim /etc/sysconfig/network-scripts/ifcfg-enp0s3 # 为228 service network restart
    rm -rf /root/hadoop-2.8.1/tmp
    echo "192.168.1.228 zookeeper4">> /etc/hosts
    echo "zookeeper4">> /root/hadoop-2.8.1/etc/hadoop/slaves 
    scp /etc/hosts root@cluster1:/etc/hosts
    scp /etc/hosts root@cluster2:/etc/hosts
    scp /etc/hosts root@cluster3:/etc/hosts
    scp /etc/hosts root@cluster4:/etc/hosts
    scp /etc/hosts root@zookeeper1:/etc/hosts
    scp /etc/hosts root@zookeeper2:/etc/hosts
    scp /etc/hosts root@zookeeper3:/etc/hosts
    scp /root/hadoop-2.8.1/etc/hadoop/slaves \
        root@cluster1:/root/hadoop-2.8.1/etc/hadoop/slaves
    scp /root/hadoop-2.8.1/etc/hadoop/slaves \
        root@cluster2:/root/hadoop-2.8.1/etc/hadoop/slaves
    scp /root/hadoop-2.8.1/etc/hadoop/slaves \
        root@cluster3:/root/hadoop-2.8.1/etc/hadoop/slaves
    scp /root/hadoop-2.8.1/etc/hadoop/slaves \
        root@cluster4:/root/hadoop-2.8.1/etc/hadoop/slaves
    scp /root/hadoop-2.8.1/etc/hadoop/slaves \
        root@zookeeper1:/root/hadoop-2.8.1/etc/hadoop/slaves
    scp /root/hadoop-2.8.1/etc/hadoop/slaves \
        root@zookeeper2:/root/hadoop-2.8.1/etc/hadoop/slaves
    scp /root/hadoop-2.8.1/etc/hadoop/slaves \
        root@zookeeper3:/root/hadoop-2.8.1/etc/hadoop/slaves
    hadoop-daemon.sh start datanode
    ll /root/hadoop-2.8.1/tmp/dfs/data/current/BP-xx\
        /current/finalized/subdir0/subdir0
    # 会自动将block副本复制到这个节点上,来保存三个副本
    # 在cluster3,cluster14
    # yarn-daemon.sh start nodemanager
    # 在cluster1,cluster12刷新集群节点
    # hdfs dfsadmin -refreshNodes

现在再将zookeeper3的datanode启动起来

1

hadoop-daemon.sh start datanode

目前将会有四个节点,四个副本,导致了数据的冗余,hadoop会自动的清除掉冗余的数据

hadoop工具(一):hive:它只是一个工具可在任意一个节点安装,和集群没有任何关系,.

安装使用hive(cluster1)

install mysql 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

# https://dev.mysql.com/downloads/repo/yum/ 查看对应版本
wget https://dev.mysql.com/get/mysql57-community-release-el7-9.noarch.rpm
rpm -ivh mysql57-community-release-el7-9.noarch.rpm
yum install -y mysql-server
echo "skip-grant-tables" >> /etc/my.cnf 
systemctl start mysqld
mysql -uroot -proot
    flush privileges;
    ALTER USER 'root'@'localhost' IDENTIFIED BY 'root';
sed -i -e '/^skip-grant-tables/d' /etc/my.cnf 
systemctl restart mysqld
mysql -uroot -proot
    CREATE DATABASE `hive` DEFAULT CHARACTER SET utf8 COLLATE utf8_general_ci;
    grant all privileges on *.* to root@localhost identified by 'root';
    flush privileges;

install hive

scp ./apache-hive-2.1.1-bin.tar.gz root@cluster1:/root

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

cd /root
tar -zxvf apache-hive-2.1.1-bin.tar.gz
mv apache-hive-2.1.1-bin hive
mv hive hadoop-2.8.1/
cd hadoop-2.8.1/hive/conf/
# cp hive-default.xml.template hive-site.xml
cp hive-env.sh.template hive-env.sh
cp hive-log4j2.properties.template hive-log4j2.properties
cp hive-exec-log4j2.properties.template hive-exec-log4j2.properties
vim hive-site.xml
    <configuration>
    <property>
        <name>javax.jdo.option.ConnectionURL</name>
        <value>jdbc:mysql://localhost:3306/hive?createDatabaseIfNotExist=true&amp;useUnicode=true&amp;characterEncoding=UTF-8&amp;useSSL=false</value>
    </property>
    <property>
        <name>javax.jdo.option.ConnectionDriverName</name>
        <value>com.mysql.jdbc.Driver</value>
    </property>
    <property>
        <name>javax.jdo.option.ConnectionUserName</name>
        <value>root</value>
    </property>
    <property>
        <name>javax.jdo.option.ConnectionPassword</name>
        <value>root</value>
    </property>
    <property>
        <name>hive.metastore.schema.verification</name>
        <value>false</value>
    </property>
    </configuration>

hive/lib目录下放mysql.jar

1

scp ./mysql-connector-java-5.1.44-bin.jar root@cluster1:/root/hadoop-2.8.1/hive/lib/

配置环境变量

1
2
3
4

echo "export HIVE_HOME=/root/hadoop-2.8.1/hive" >> ~/.bashrc
echo "export PATH=\${PATH}:\${HIVE_HOME}/bin" >> ~/.bashrc
echo "export HIVE_CONF_DIR=\${HIVE_HOME}/conf" >> ~/.bashrc
source ~/.bashrc

启动hive,建立在集群之上

1
2
3
4
5
6
7
8
9
10
11

# 启动三台datanode(ZK)
zkServer.sh start 
hadoop-daemon.sh start journalnode
# 启动start-dfs.sh
start-dfs.sh 
# hive在第一次登录的时候需要用如下命令初始化    
schematool -dbType mysql -initSchema
# 启动hive metastore进程
hive --service metastore & 
# 运行hive
hive

创建一些数据

1
2
3
4
5
6

vim /root/d.data
1 iphone8 64G 8000
2 iphone7 64G 7000
3 iphone6 64G 6000
4 iphone5 64G 5000
5 iphone4 64G 4000

hive语法

1
2
3
4
5
6
7
8
9
10
11
12
13

SHOW TABLES;
create table t_order(id int,name string,container string,price double)
row format delimited 
fields terminated by ' ';
SHOW TABLES;
# 数据本来只那里,只能用来查看
# 所以不能使用 insert into **
# 向表中加载数据
load data local inpath '/root/d.data' into table t_order;
# 查看与统计
select * from t_order;
select count(*) from t_order; # 执行的MapReduce程序

通过mysql查看hive上传的时的表的信息

1
2
3
4
5
6
7
8
9
10
11

user hive;
show tables;
# 记录元数据的表
select * from DBS;
# DB_LOCATION_URI: hdfs://ns1/user/hive/warehouse
# 查看数据库的内容 http://cluster1:50070/explorer.html#/user/hive/warehouse
# 查看在hive中创建的表
select * from TBLS;
# 表中的字段
select * from COLUMNS_V2;

用hive创建数据库或表就相当于在hdfs中创建目录

Zookeeper是Google的Chubby的一个开源的实现,是Hadoop的分布协调服务,它包含一个简单的原语集,分布式应用程序可以基于它实现同步服务,配置维护和命名服务等,保证hadoop的HA(高可用),它本身也是个集群(提供少量数据的存储和管理)

大部分分布式应用需要一个主控,协调器或控制器来管理物理分布的子进程(资源,任务分配等)
Zookeeper:提供通用的分布式锁服务,用以协调分布式应用

Zookeeper安装和配置(集群模式)
在一台虚拟机上运行如下脚本

1

curl https://file.femnyy.com/file/install_zookeeper.sh |sudo sh

准备工作

1. clone三台之后,修改主机名为zookeeper1,zookeeper2,zookeeper3,修改/etc/hosts
2. 修改IP /etc/sysconfig/network-scripts/ 在的三个IP
3. 关闭防火墙
4. ssh免登陆
5. 安装JDK,配置环境变量
6. zoo.conf的配置dataDir中的目录下创建一个myid文件
   再修改vim /root/zookeeper-3.4.10/data/myid 对应的Id号
   启动Zookeeper:

   1 2	# 在三台虚拟机中都执行 zkServer.sh start

查看Zookeeper状态:

1
2

# 在三台虚拟机中都执行
zkServer.sh status

提供一个命令行的客户端

1
2
3
4
5
6
7
8
9

zkCli.sh
help
# 创建一个共同的配置文件,所有的zookeeper集群都能访问到
create /zp241 path:/home/femn/ 
get /zp241 
# 启动另一个虚拟机的命令行客户端也可以看到上传的内容
zkCli.sh
get /zp241 
quit

zookeeper管理客户所存放的数据采用的是类似于文件树的结构
每一个节点叫做一个node:节点分为两种类型短暂的和持久
短暂的Node的客户端会话结束时,zookeeper会将该短暂的node删除,同时 短暂的node不可以有子节点

hadoop运行jar包

运行一个mapreduce程序job(打包成一个jar包)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

hadoop jar 
cd /root/hadoop-2.8.1/share/hadoop/mapreduce
# 用mapreduce计算圆周率 map的任务数量 每一map的取样数
hadoop jar hadoop-mapreduce-examples-2.8.1.jar pi 5 5
# yarn 创建了tmp ,同时mapreduce程序创建了user目录
# 运行计算单词出现的次数
# wordcount 将此目录下的所有文件进行统计 结果输出到此目录下
vim test.txt
    hello world
    hello femn
    hello leipengkai
    hello friend
hadoop fs -put test.txt /wordcount/input
hadoop jar hadoop-mapreduce-examples-2.8.1.jar wordcount \
    /wordcount/input /wordcount/output

MapReduce

处理海量数据的运算, 即使是一个很简单的逻辑,要把它变成分布式运行的程序将会面临很多的其它问题:

运算代码程序的资源分发和启动程序的环境配置,以及代码分发到哪些datanode上,并且还得监控datanode运行状态是否正常 以及datanode的调试汇总

解决思路:运算往数据方移动,而不是数据移动到运算这方来.

MapReduce分成两个步骤去完成业务逻辑:
Map逻辑和Reduce逻辑都会在分布在datanode中,先执行Map,再执行Reduce.在Map程序时可高并发执行

实例代码

vim WCMAP.java

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

package cn.itcast.hadoop.wcmapreduce;
import java.io.IOException;
import java.util.StringTokenizer;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.util.StringUtils;
// 4个泛型中,前两个是指定map输入数据类型,KEY是输入的key类型,
// VALUEIN是输入的value的类型,后面两个是map输出给reduce的输出数据类型
// 默认情况下,框架传递给我们的map的输入数据中,
// key是要处理的文本(block)中一行的起始偏移量,这一行的内容作为value
public class WCMap extends Mapper<LongWritable, Text, Text, LongWritable>{
// Long,String,String,Long等内存对象 经过序列化之后再通过网络传递到节点中
// hadoop实现了自己的序列化机制 去掉多余的java序列化信息	
    
    private final static LongWritable one =new LongWritable(1);
    private Text word = new Text();
    // map()框架第每读一行数据就调用一次该方法,
    // 对节点中的文本处理完所有的map之后才进入reduce
    protected void map(LongWritable key,Text value,Context context)
            throws IOException, InterruptedException{
        // 具体业务逻辑就写在这个方法体中,而且我们业务要处理的数据已经被框架传递进来
// 在方法的参数中key-value中
        // key是这一行数据的起始偏移量 value是这一行的文本内容 
        // context传到reduce的工具不用自己去找节点
        String line = value.toString();
        char split = ' ';
        String[] words =StringUtils.split(line, split);
        //
        //遍历这个单词数组输出为k-v形式 k:单词 v:1
        for (String word: words) {
            context.write(new Text(word), new LongWritable(1));
        }
//	 StringTokenizer itr = new StringTokenizer(value.toString());
//	 while(itr.hasMoreTokens()) {
//		 word.set(itr.nextToken());
//		 context.write(word, one);
//		 }
    }
}

vim WCReduce.java

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

package cn.itcast.hadoop.wcmapreduce;
import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
public class WCReduce extends Reducer<Text, LongWritable, Text, LongWritable> {
    // 框架在map处理完之后,将所有的k-v缓存起来,进行分级,然后传递一个组,<k,vs{}>
    //调用一次reduce方法  <hello,{1,1...}>
    @Override
    protected void reduce(Text key, Iterable<LongWritable> values,Context context) 
                    throws IOException, InterruptedException {
        // TODO Auto-generated method stub
//		super.reduce(key, values, context);
        long count = 0;
        for (LongWritable v:values) {
            count += v.get();
        }
        context.write(key, new LongWritable(count));
    }
}

vim WCRunner.java

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
49
50

package cn.itcast.hadoop.wcmapreduce;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
//import org.apache.hadoop.mapred.jobcontrol.Job;
public class WCRunner {
    //用来描述一个特定的作业
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        // TODO Auto-generated method stub
        Configuration conf = new Configuration();
        conf.set("mapreduce.job.jar","/root/wc.jar");    
        Job job = Job.getInstance(conf);
        //设置整个job手忙脚乱的那些类在哪个jar包中
        job.setJarByClass(WCRunner.class);
        //该作业使用哪个类作为逻辑处理中的map,哪个作为reduce
        job.setMapperClass(WCMap.class);
        job.setReducerClass(WCReduce.class);
        //指定map输出数据类型k-v
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);
        //指定reduce输出数据类型k-v
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);
        //该作业要处理的数据所在的路径以及输出结果存放路径
        // hdfs只有一个test.txt文件
        FileInputFormat.setInputPaths(job, new Path("hdfs://cluster1:9000/"));
        FileOutputFormat.setOutputPath(job, new Path("hdfs://cluster1:9000/wc/output"));
        
        //提交集群运行
        //yarn机制
        job.waitForCompletion(true);
    }
}

在本地测试可以启动Bebug模式,直接在虚拟机下用eclipse运行WCRunner

注意:如果是eclipse它会自动加载core-site.xml和hdfs-site.xml的配置文件,
所以可以写成如下,也是读取hdfs文件系统

1
2

FileInputFormat.setInputPaths(job, new Path("/"));
FileOutputFormat.setOutputPath(job, new Path("/wc/output"));

使用dhfs的话,可以不用启动yarn,只启动dhfs,这时的Mapreduce程序将跑在本机
不通过yarn分配

如果想要yarn分配在Node Manage运行Mapreduce,则在src下加上mapred-site.xml和yarn-site.xml文件
并加上如下代码,再运行

1

conf.set("mapreduce.job.jar","/root/wc.jar");

也可以不用从hdfs中读写文件:

1
2

FileInputFormat.setInputPaths(job, new Path("/root/test.txt"));
FileOutputFormat.setOutputPath(job, new Path("/root/output"));

在集群中运行,将将整个项目打包成jar,让hadoop yarn分发运行Mapreduce程序

1
2
3
4

hadoop fs -rm -f -R hdfs://cluster1:9000/*
hadoop fs -put test.txt hdfs://cluster1:9000/
hadoop jar wc.jar cn.itcast.hadoop.wcmapreduce.WCRunner 
# -Dmapreduce.input.fileinputformat.input.dir.recursive=true

注意:如果使用jar包运行的话,必须得指定成hdfs的文件系统格式

1
2

FileInputFormat.setInputPaths(job, new Path("hdfs://cluster1:9000/"));
FileOutputFormat.setOutputPath(job, new Path("hdfs://cluster1:9000/wc/output"));

查看运行的结果

1

hadoop fs -cat  hdfs://cluster1:9000/wc/output/part-r-00000

将信息打印到控制台

1

export HADOOP_ROOT_LOGGER=DEBUG,console

总结MR程序的提交运行模式

本地模式

在eclipse里面直接运行main方法,不添加mapred-site.xml和yarn-site.xml
也会提交给localjobnumber执行,输出输出数据按如下设置保存在对应路径下:

1
2

FileInputFormat.setInputPaths(job, new Path("/"));
FileOutputFormat.setOutputPath(job, new Path("/wc/output"));

集群模式运行

1. 将项目打成jar包,上传到服务器上,然后用hadoop命令提交:

   1	hadoop jar wc.jar cn.itcast.hadoop.wcmapreduce.WCRunner

2. 在eclipse中运行main,在src下加上mapred-site.xml和yarn-site.xml文件
   并加上如下代码,再运行

   1	conf.set("mapreduce.job.jar","/root/wc.jar");

Map进程数不是由block大小决定的,而是由一个切片(split)对应一个Map进程
Map task的并发数量是由切片的数量决定的,有多少个切片就启动多少个Map task

切片是个逻辑概念,指文件中数据中偏移量范围,block是物理概念
切片的具体大小应该根据所处理的文件大小来调整

RPC

远程过程调用(RPC):hadoop的节点之间的进程通信(类与类之间),以及与客户端的通信.心跳也是通过RPC完成的
为什么datanode定期会向Namenode汇报block信息

RPC的现实技术:动态代理,反射,socker通信生成RPC客户端

serveice

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

package cn.itcast.hadoop.hdfs;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.ipc.RPC.Builder;
import org.apache.hadoop.ipc.RPC.Server;
publci class ServiceStart{
    public static void main(String[] args) 
        throws HadoopIllegalArgumentException, IOException {
        Builder builder = new RPC.Builder(new Configuration());
        builder.setBindAddress("cluster1").setPort(10000)
            .setProtocol(LoginInter.class).setInstance(new LoginServer());
        Server server = builder.build();
        server.start();
        // 多个业务逻辑处理
    }
}

1
2
3
4
5
6
7
8

package cn.itcast.hadoop.hdfs;
public class LoginServer implements LoginInter {
    @Override
    public String login(String username, String passwd) {
        return username + "success!!";
    }
}

client

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

package cn.itcast.hadoop.hdfs;
import java.io.IOException;
import java.net.InetSocketAddress;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.ipc.RPC;
public class LoginControl {
    public static void main(String[] args) throws IOException {
        // TODO Auto-generated method stub
        LoginInter proxy = RPC.getProxy(LoginInter.class, 1L,
         new InetSocketAddress("192.168.1.222",10000), new Configuration());
        String r = proxy.login("femn", "femn");
        System.out.println(r);
        // 当增加服务器提供服务时,客户端能够感知服务端.
        // 在不改变的情况下,知道去调用新服务器的服务
        // 服务调用动态转发和负载均衡的实现,使用zookeeper可以很容易的实现
    }
}

service and client

1
2
3
4
5
6

package cn.itcast.hadoop.hdfs;
public interface LoginInter {
    public static final long versionId=1L;
    public String login(String username,String passwd);
}