分类： Storm

项目架构

基本要求

1. 主题相关的WebAPP
   1. 1个首页和若干功能页面
   2. 有生成数据的能力（实际展示的时候，为了有更多丰富的数据, 允许使用脚本生成假数据）
      1. KafkaProducer 直接将需要采集的信息发送到Storm
   3. 分析结果的图表展示
      1. echarts, 或者其他图表库
         1. Jquery的ajax库
         2. json
   4. 其他要求：
      1. 参照Alibaba的Java开发手册的规约
      2. 代码中适当添加注释
2. 部署到服务器（虚拟机或者购买的服务器）
   1. JavaWeb程序需要部署到 Tomcat
   2. Nginx 实现反向代理
3. 实时数据分析
   1. kafka -> storm
   2. 持久化到数据库（HBase，MySQL，Redis）
   3. 最好每个团队成员都有一个完整的实时分析流程
   4. 每个团队最少有两种实时分析, 不同类型（不要过于简单，分析的内容要能体现实时性）

开发流程建议

1. 确定需求和分工

2. 学习使用版本控制工具(Gitee 码云 / Github)

3. 统一消息格式，建议使用JSON

4. 开发时先采用本地的Tomcat和Storm环境测试

5. 本地环境测试时可以远程连接服务器上的的Kafka和Hbase

   没问题再使用Tomcat和Nginx部署到服务器

6. 将拓扑上传到Storm集群中运行

7. 联调所有模块

8. 完善SRS报告

   记录整个项目的需求, 架构, 开发、部署、测试的详细过程

9. 每个人准备PPT, 视频等交付资料

10. 每周使用在线表格记录项目进度情况

WebApp开发

采用前后端分离的模式开发和部署

后端可以采用基于tomcat的普通JavaWeb应用或者SpringBoot项目

消息传递

一般项目可以使用kafka-clients依赖里的KafkaProducer API来发送消息到Kafka, 这样便于快速调试.

对于SpringBoot项目,可以使用spring-kafka依赖.

kafka相关配置

spring:
  kafka:
    bootstrap-servers: hadoop000:9092,hadoop000:9093,hadoop000:9094
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      client-id: app-pro-cli
      acks: 1
      retries: 3
    consumer:
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      client-id: app-pro-cli
      group-id: g1

Kafka的初始化配置

package cn.delucia.project.conf;

import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class KafkaInitConf {
    @Value("${app.kafka-topic}")
    private String kafkaTopic;
    @Value("${app.topic-partitions}")
    private Integer topicPartitions;
    // 创建一个Topic并设置分区数和副本数
    @Bean
    public NewTopic initialTopic() {
        return new NewTopic(kafkaTopic, 1, (short) 1);
    }
    // 如果要修改分区数，只需修改配置值重启项目即可
    // 修改分区数并不会导致数据的丢失，但是分区数只能增大不能减小
    @Bean
    public NewTopic updateTopic() {
        return new NewTopic(kafkaTopic, topicPartitions, (short) 1);
    }
}

在控制器类中注入kafkaTemplate并调用send方法发送消息, 例如:

@Slf4j
@RestController
public class GreetingController {

    @Autowired
    private KafkaTemplate<Object, String> kafkaTemplate;

    public Greeting greeting(@RequestParam(value = "name", defaultValue = "农场主") String name) {

        Greeting greeting = new Greeting(counter.incrementAndGet(), name);

        try {
            String s = new ObjectMapper().writeValueAsString(greeting);
            // 带回调的生产者
            kafkaTemplate.send(kafkaTopic, "Greeting:" + s).addCallback(success -> {
                // 消息发送到的topic
                String topic = Objects.requireNonNull(success).getRecordMetadata().topic();
                // 消息发送到的分区
                int partition = success.getRecordMetadata().partition();
                // 消息在分区内的offset
                long offset = success.getRecordMetadata().offset();
                log.info("发送消息成功: {}-{}-{}, Greeting:{}", topic, partition, offset, s);
            }, failure -> {
                log.info("发送消息失败: {}", failure.getMessage());
            });
        } catch (JsonProcessingException e) {
            log.info("解析Json格式失败: {}", e.getMessage());
        }
        ...
    }
}

安装Tomcat

准备：安装Java JDK1.8

这里安装openjdk是因为比较简单, 工作场合一定要使用oracle提供的jdk1.8

sudo yum -y install java-1.8.0-openjdk*

这样安装的好处就是环境变量都配好了

可以直接查看版本 java -version

Tomcat安装

下载页面： https://tomcat.apache.org/download-90.cgi

文档：https://tomcat.apache.org/tomcat-9.0-doc/index.html

下载链接

wget https://mirrors.tuna.tsinghua.edu.cn/apache/tomcat/tomcat-9/v9.0.41/bin/apache-tomcat-9.0.41.tar.gz

解压到后改名tomcat9

tar -zxvf apache-tomcat-9.0.41.tar.gz -C ~/app
cd ~/app
mv apache-tomcat-9.0.41/ tomcat9

默认tomcat端口是8080, 为了避免冲突, 这里修改为18080

 $ vi ~/app/tomcat9/conf/server.xml

 69     <Connector port="18080" protocol="HTTP/1.1"
 70                connectionTimeout="20000"
 71                redirectPort="8443" />

启动: 进入 /bin目录下 运行startup.sh脚本文件

$ ./startup.sh 
Using CATALINA_BASE:   /home/hadoop/app/tomcat9
Using CATALINA_HOME:   /home/hadoop/app/tomcat9
Using CATALINA_TMPDIR: /home/hadoop/app/tomcat9/temp
Using JRE_HOME:        /home/hadoop/app/jdk1.8.0_211
Using CLASSPATH:       /home/hadoop/app/tomcat9/bin/bootstrap.jar:/home/hadoop/app/tomcat9/bin/tomcat-juli.jar
Tomcat started.

检查tomcat进程信息

$ bin]$ ps -ef | grep tomcat
hadoop    15142      1  1 14:41 pts/0    00:00:05 /home/hadoop/app/jdk1.8.0_211/bin/java -Djava.util.logging.config.file=/home/hadoop/app/tomcat9/conf/logging.properties -Djava.util.logging.manager=org.apache.juli.ClassLoaderLogManager -Djdk.tls.ephemeralDHKeySize=2048 -Djava.protocol.handler.pkgs=org.apache.catalina.webresources -Dorg.apache.catalina.security.SecurityListener.UMASK=0027 -Dignore.endorsed.dirs= -classpath /home/hadoop/app/tomcat9/bin/bootstrap.jar:/home/hadoop/app/tomcat9/bin/tomcat-juli.jar -Dcatalina.base=/home/hadoop/app/tomcat9 -Dcatalina.home=/home/hadoop/app/tomcat9 -Djava.io.tmpdir=/home/hadoop/app/tomcat9/temp org.apache.catalina.startup.Bootstrap start
hadoop    15296  15064  0 14:47 pts/0    00:00:00 grep --color=auto tomcat

检查对应的监听端口信息

netstat -anpt | grep 15142

网页访问虚拟机主机名或IP地址:18080

为了方便可以为tomcat的安装目录配置环境变量CATALINA_HOME,添加到PATH中

为tomcat添加用户和角色

修改conf/tomcat-users.xml, 添加如下内容

<!--  
  <role rolename="tomcat"/>
  <role rolename="role1"/>
  <user username="tomcat" password="tomcat" roles="tomcat"/>
  <user username="both" password="tomcat" roles="tomcat,role1"/>
  <user username="role1" password="tomcat" roles="role1"/>
-->
  <role rolename="manager-gui"/>
  <role rolename="manager-script" />
  <user username="admin" password="123123" roles="manager-gui,manager-script" />
</tomcat-users>

修改webapps/manager/META-INF目录下的context.xml,在allow行的末尾加上|\d+.\d+.\d+.\d+表示允许所有主机访问。

<Context antiResourceLocking="false" privileged="true" >
  <Valve className="org.apache.catalina.valves.RemoteAddrValve"
         allow="127\.\d+\.\d+\.\d+|::1|0:0:0:0:0:0:0:1|\d+\.\d+\.\d+\.\d+" />
  <Manager sessionAttributeValueClassNameFilter="java\.lang\.(?:Boolean|Integer|Long|Number|String)|org\.apache\.catalina\.filters\.CsrfPreventionFilter\$LruCache(?:\$1)?|java\.util\.(?:Linked)?HashMap"/>
</Context>

重启tomcat生效

部署项目

一般的javaee项目可以直接build出一个war包进行上传服务器。

SpringBoot默认是打成jar包，如果需要打成war包， 需要修改pom.xml文件：

    <groupId>com.niit</groupId>
    <artifactId>demo</artifactId>
    <version>0.0.1-SNAPSHOT</version>

    <!-- 这里打成war包 若打jar，需将war改为jar -->
    <packaging>war</packaging>

    <name>demo</name>
    <description>Demo project for Spring Boot</description>

然后使用mvn:package构建war包即可，然后上传到服务器的tomcat目录下的webapp文件夹之内。

tomcat 容器的运行机制👇

tomcat默认会加载tomcat目录下的webapp文件夹之内的文件，如

其中ROOT目录下为Tomcat的欢迎页

http://hadoop000/index.jsp

http://hadoop000/tomcat.gif

examples目录是一些官方示例

http://hadoop000/examples/

tomcat也会默认会加载tomcat目录下的webapp文件夹中下面的war包，并自动解压在webapp下面。

默认的访问方式就是 http://域名:端口号/war包名, 端口号默认是8080

启动tomcat， 浏览器访问 http://hadoop000:18080/demo/

部署SpringBoot项目

项目的服务器设置:

server:
  port: 18080
  servlet:
    context-path: "/project"
#debug: on

使用maven的springboot打包插件按照jar包的方式打包

   <groupId>com.niit</groupId>
    <artifactId>demo</artifactId>
    <version>0.0.1-SNAPSHOT</version>

    <!-- 这里打成war包 若打jar，需将war改为jar -->
    <packaging>jar</packaging>

    <name>demo</name>
    <description>Demo project for Spring Boot</description>

然后使用mvn:package打包即可，将打包出来的jar文件重命名然后上传到服务器

在服务器启动项目（需要Java8以上环境）

[hadoop@hadoop000 webapps]$ java -jar demo-jar.jar 

  .   ____          _            __ _ _
 /\\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \
( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
 \\/  ___)| |_)| | | | | || (_| |  ) ) ) )
  '  |____| .__|_| |_|_| |_\__, | / / / /
 =========|_|==============|___/=/_/_/_/
 :: Spring Boot ::        (v2.3.0.RELEASE)

如果后台运行可以(推荐)

$ nohup java -jar demo-jar.jar 1> demo.log 2>&1 &

浏览器访问 http://hadoop000:18080/project/greeting

配置Nginx反向代理

但是这样带8080端口的访问并不是很好，因此一般都使用nginx反向代理.

通过反向代理将对nginx的80端口的访问请求转发到tomcat的8080端口

首先在物理机hosts文件创建虚拟机主机名和ip的映射

192.168.186.100 hadoop000

Nginx安装

安装前确认是否已经安装过

sudo yum search nginx

Nginx文档： http://nginx.org/en/docs/

Installation on Linux, nginx packages from nginx.org can be used.

Installation instructions
RHEL/CentOS
Debian
Ubuntu
SLES
Alpine

以 CentOS为例进行安装

Install the prerequisites:

sudo yum install yum-utils

To set up the yum repository, create the file named /etc/yum.repos.d/nginx.repo with the following contents:

sudo vi /etc/yum.repos.d/nginx.repo

内容如下：

[nginx-stable]
name=nginx stable repo
baseurl=http://nginx.org/packages/centos/$releasever/$basearch/
gpgcheck=1
enabled=1
gpgkey=https://nginx.org/keys/nginx_signing.key
module_hotfixes=true

[nginx-mainline]
name=nginx mainline repo
baseurl=http://nginx.org/packages/mainline/centos/$releasever/$basearch/
gpgcheck=1
enabled=0
gpgkey=https://nginx.org/keys/nginx_signing.key
module_hotfixes=true

By default, the repository for stable nginx packages is used. If you would like to use mainline nginx packages, run the following command:

sudo yum-config-manager --enable nginx-mainline

To install nginx, run the following command:

sudo yum install -y nginx

When prompted to accept the GPG key, verify that the fingerprint matches 573B FD6B 3D8F BC64 1079 A6AB ABF5 BD82 7BD9 BF62, and if so, accept it.

Nginx配置

创建java.conf ，进行最简配置

$ cd /etc/nginx/conf.d/
$ sudo cp default.conf java.conf
$ vi java.conf

URL记住不要忘了加http://前缀

server {
    listen  80;
    server_name hadoop000;
    location / {
        proxy_pass http://127.0.0.1:18080;
    }
}

常用命令

解释	命令
安装服务	yum install nginx
启动服务	service nginx start
停止服务	service nginx stop
重载服务	service nginx reload

配置完成后启动服务

sudo service nginx start

如果服务器已经启动，当配置发生变化可以直接使用重载服务来更新配置，运维常用，因为不需要停止服务就可以重载新的配置。

如果启动失败可以查看错误日志

sudo vi /var/log/nginx/error.log 

经过反向代理配置之后，使用浏览器访问 http://hadoop000就相当于访问虚拟机hadoop000的本地服务http://127.0.0.1:18080的效果

如果发现502错误：

2020/11/18 16:12:39 [crit] 16376#16376: *1 connect() to 127.0.0.1:8080 failed (13: Permission denied) while connecting to upstream, client: 192.168.186.1, server: hadoop000, request: "GET /favicon.ico HTTP/1.1", upstream: "http://127.0.0.1:8080/favicon.ico", host: "hadoop000", referrer: "http://hadoop000/"

此时需要考虑把linux操作系统默认的强制访问安全限制设置为禁用。

关闭SElinux即可

1. 临时关闭 SElinux

   sudo setenforce 0

2. 永久关闭 SElinux

   sudo vim /etc/selinux/config
   SELINUX=disabled

修改之后就可以正常访问了

配置开机启动

[hadoop@hadoop000 download]$ chkconfig nginx
注意：正在将请求转发到“systemctl is-enabled nginx.service”。
disabled

[hadoop@hadoop000 download]$ chkconfig nginx on
注意：正在将请求转发到“systemctl enable nginx.service”。
==== AUTHENTICATING FOR org.freedesktop.systemd1.manage-unit-files ===
Authentication is required to manage system service or unit files.
Authenticating as: root
Password: 
==== AUTHENTICATION COMPLETE ===
Created symlink from /etc/systemd/system/multi-user.target.wants/nginx.service to /usr/lib/systemd/system/nginx.service.
==== AUTHENTICATING FOR org.freedesktop.systemd1.reload-daemon ===
Authentication is required to reload the systemd state.
Authenticating as: root
Password: 
==== AUTHENTICATION COMPLETE ===

[hadoop@hadoop000 download]$ chkconfig nginx
注意：正在将请求转发到“systemctl is-enabled nginx.service”。
enabled

前后端分离

nginx是一个高性能服务器，除了配置反向代理之外，也非常适合部署静态资源并支持高并发请求, 并且也提供负载均衡的功能。

首先删除默认的配置

# sudo vim /etc/nginx/conf.d/default.conf

为了将反向代理请求和静态资源的请求分开，修改我们之前的配置如下：

upstream webapp.server {
    server hadoop000:18080;
}

server {
        listen  80;
        server_name localhost hadoop000;
        root /data/www/;

        # 静态资源
        location / {
            index index.html;
            access_log /var/log/nginx/java-hadoop.log main;
        }

        # 反向代理到本地JavaWeb的后台服务
        location ^~ /project/ {
            proxy_pass http://webapp.server/project/;
        }
}

一些说明如下：

1. listen 80 是http协议的默认端口，:80 可以省略
2. server_name 表示请求路径中的服务器主机名，可配置多个
3. /data/www 为站点根目录，需手动创建，权限一般为755
4. access_log 对应的路径是应用的服务器日志，文件夹不存在则需手动创建
5. location 的匹配规则，优先匹配 /demo/，其次 /

一般来说前后端分离部署应该是部署在不同的服务器上的，这里放在一台服务器上只是为了演示方便。

前端部署

然后将项目的静态页面放到/data/www下即可

后端部署

SpringBoot项目移除静态资源，单独部署在tomcat上，并使用ngixn实现反向代理

实时数据采集

主要技术：Kafka

实时数据流计算

主要技术：Kafka-clients，storm-hbase|storm-redis|storm-mysql

以热力图项目为例, 拓扑代码如下

package com.niit.project;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.hbase.bolt.HBaseBolt;
import org.apache.storm.hbase.bolt.mapper.SimpleHBaseMapper;
import org.apache.storm.kafka.spout.ByTopicRecordTranslator;
import org.apache.storm.kafka.spout.KafkaSpout;
import org.apache.storm.kafka.spout.KafkaSpoutConfig;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.HashMap;
import java.util.Map;

public class KafkaStormProjectApp {

    public static String topologyName = "project-topo";
    public static final String KAFKA_BROKER = "hadoop000:9092";
    public static final String INPUT_TOPIC = "storm-project";

    private static class SplitBolt extends BaseRichBolt {
        private OutputCollector outputCollector;

        @Override
        public void prepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector collector) {
            this.outputCollector = collector;
        }

        @Override
        public void execute(Tuple input) {
            String id = input.getStringByField("line");
            String[] split = id.split(",");

            try {
                double lng = Double.parseDouble(split[0]);
                double lat = Double.parseDouble(split[1]);
                this.outputCollector.emit(new Values(id, lng, lat));
            } catch (NumberFormatException e) {
                System.err.println(e.getMessage());
            }
        }

        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            declarer.declare(new Fields("id", "lng", "lat"));
        }
    }

    /**
     * 计数利用HBase的CountColumn特性
     */
    private static class CountBolt extends BaseRichBolt {

        private OutputCollector collector;
        private final HashMap<String, Long> counts = null;

        @Override
        public void prepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector collector) {
            this.collector = collector;
        }

        @Override
        public void execute(Tuple input) {
            String id = input.getString(0);
            double lng = input.getDouble(1);
            double lat = input.getDouble(2);

            this.collector.emit(new Values(id, lng, lat, 1L));
        }

        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            declarer.declare(new Fields("id", "lng", "lat", "count"));
        }
    }

    public static void main(String[] args) throws Exception {

        final TopologyBuilder builder = new TopologyBuilder();

        // storm conf
        Config conf = new Config();
        conf.setNumAckers(0);
        conf.setDebug(true);

        // kafka bolt
        ByTopicRecordTranslator<String, String> translator =
                new ByTopicRecordTranslator<>((r) -> new Values(r.value()), new Fields("line"));
        translator.forTopic(INPUT_TOPIC, (r) -> new Values(r.value()), new Fields("line"));

        KafkaSpoutConfig<String, String> kafkaSpoutConfig = KafkaSpoutConfig
                // bootstrapServers 以及topic
                .builder(KAFKA_BROKER, INPUT_TOPIC)
                // 设置group.id
                .setProp(ConsumerConfig.GROUP_ID_CONFIG, "location")
                // ensure at-least-once processing
                .setProp(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
                // 设置开始消费的起始位置
                // 设置提交消费边界的时长间隔
                .setOffsetCommitPeriodMs(10_000)
                //Translator
                .setRecordTranslator(translator)
                .build();

        KafkaSpout<String, String> kafkaSpout = new KafkaSpout<>(kafkaSpoutConfig);

        // hbase bolt
        Map<String, Object> hbConf = new HashMap<>();
        hbConf.put("hbase.rootdir", "hdfs://hadoop000:9000/hbase");
        hbConf.put("hbase.zookeeper.quorum", "hadoop000:2181");
        conf.put("hbase.conf", hbConf);
        conf.setNumWorkers(2);  // 设置为1个topology创建2个worker进程

        SimpleHBaseMapper mapper = new SimpleHBaseMapper()
                .withRowKeyField("id")
                .withColumnFields(new Fields("lng","lat"))
                .withCounterFields(new Fields("count"))
                .withColumnFamily("cf");

        HBaseBolt hbaseBolt = new HBaseBolt("project", mapper).withConfigKey("hbase.conf");

        // build topology
        builder.setSpout("kafka_spout", kafkaSpout);
        builder.setBolt("split-bolt", new SplitBolt(), 2)
                .setNumTasks(4)
                .shuffleGrouping("kafka_spout");
        builder.setBolt("count-bolt", new CountBolt())
                .fieldsGrouping("split-bolt", new Fields("id"));
        builder.setBolt("hbase-bolt", hbaseBolt).globalGrouping("count-bolt");

        if (args != null && args.length > 0) {
            topologyName = args[0];
            StormSubmitter.submitTopology(topologyName, conf, builder.createTopology());
        } else {
            LocalCluster localCluster = new LocalCluster();
            localCluster.submitTopology(topologyName, conf, builder.createTopology());
        }
    }

}

其中聚合操作是利用了HBase的CounterColumn特性

这里没有使用时间窗口Bolt来体现实时，而是利用了HBase表的TTL属性，TTL可以在创建表的时候指定，TTL设置为60即表示表中记录的存活时间为1分钟：

create "project",{NAME => 'cf', MIN_VERSIONS => '0',TTL => '60'}

也可以disable表之后使用alter语句对已有表进行修改。

由于插入数据的时候经纬度是采用了Double类型，而计数列采用了Long类型，但是HBase只有使用字节数组这样一种方式进行存储，所以需要程序员自己控制数据类型的转换。在查看表中记录的使用需要这样进行数据类型的转换：

scan 'project', {COLUMNS => ['cf:lng:toDouble','cf:lat:toDouble','cf:count:toLong']}

将上传到Storm集群

[kafka-storm-project]$ storm jar project-topology.jar com.niit.demo.KafkaStormProjectTopology project-topo

数据可视化

主要技术：百度echarts图表，异步请求图表渲染（Ajax & JSON）

这个热力图项目的前端比较简单，只有一个index.html文件

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>百度地图</title>
    <style>
        #main {
            position: absolute;
            top: 0;
            left: 0;
            right: 0;
            bottom: 0;
            width: 100%;
            height: 100%;
        }
    </style>
</head>
<body>

<!-- 为ECharts准备一个具备大小（宽高）的Dom -->
<div id="main"></div>

<!-- jquery 1.11.3 -->
<script type="text/javascript" src="https://cdn.jsdelivr.net/npm/jquery@1.11.3/dist/jquery.min.js"></script>
<!-- bootstrap 3.3.7-->
<script type="text/javascript" src="https://cdn.jsdelivr.net/npm/bootstrap@3.3.7/dist/js/bootstrap.min.js"></script>
<!-- echarts 插件  -->
<script type="text/javascript" src="https://cdn.jsdelivr.net/npm/echarts/dist/echarts.min.js"></script>
<!-- echarts 百度地图插件 -->
<script type="text/javascript" src="https://cdn.jsdelivr.net/npm/echarts/dist/extension/bmap.js"></script>
<script type="text/javascript"
        src="https://api.map.baidu.com/api?v=2.0&ak=KOmVjPVUAey1G2E8zNhPiuQ6QiEmAwZu&__ec_v__=20190126"></script>
<script>

    var points = [];
    var myChart = echarts.init(document.getElementById('main'));

    myChart.setOption(option = {
        animation: false,
        bmap: {
            center: [110.337731, 20.064295],  // 海南大学
            zoom: 18, // 地图缩放等级
            roam: true
        },
        visualMap: {
            show: false,
            top: 'top',
            min: 0,
            max: 5,
            seriesIndex: 0,
            calculable: true,
            inRange: {
                color: ['blue', 'blue', 'green', 'yellow', 'red']
            }
        },
        series: [{
            type: 'heatmap',
            coordinateSystem: 'bmap',
            data: points,
            pointSize: 5,
            blurSize: 6
        }]
    });
    // 添加百度地图插件
    var bmap = myChart.getModel().getComponent('bmap').getBMap();
    bmap.addControl(new BMap.MapTypeControl());
    // 禁止拖拽和缩放
    bmap.disableDragging();
    bmap.disableScrollWheelZoom();

    bmap.addEventListener("click", function (e) {
        $.post('log', {
            lng: e.point.lng,
            lat: e.point.lat
        });
    });

    // 10秒更新一次地图
    window.setInterval(function () {
        $.get(
            "points",
            function (data) {
                for (var i = 0; i < data.length; i++) {
                        point = [];
                    for (var j = 0; j < data[i].count; j++) {
                        points.push([data[i].lng, data[i].lat, 1]);
                    }
                }
                option.series.data = points;
                myChart.setOption(option);
            }
        )
    }, 10000);

</script>
</body>
</html>

其中请求demo/points可以到达后端的SpringBoot项目，并且返回HBase的最新数据，对应的接口如下:

src\main\java\com\niit\demo\service\LocationService.java

package com.niit.demo.service;

import com.niit.demo.entity.Point;
import com.niit.demo.utils.HBaseHelper;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;

@Component
public class LocationService {

    public List<Point> getLocationList() {

        List<Point> list = new ArrayList<>();
        ResultScanner scanner = HBaseHelper.getScanner("project");

        if (scanner != null) {
            scanner.forEach(rowResult -> {

                Point point = new Point();
                for (Cell cell : rowResult.listCells()) {

                    String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                    switch (qualifier) {
                        case "lng":
                            point.setLongitude(Bytes.toDouble(CellUtil.cloneValue(cell)));
                            break;
                        case "lat":
                            point.setLatitude(Bytes.toDouble(CellUtil.cloneValue(cell)));
                            break;
                        case "count":
                            point.setCount(Bytes.toLong(CellUtil.cloneValue(cell)));
                            break;
                        default:
                            break;
                    }
                }
                list.add(point);
            });
        }
        return list;
    }

}

src\main\java\com\niit\demo\IndexController.java

package com.niit.demo;

import com.niit.demo.entity.Point;
import com.niit.demo.service.LocationService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.List;

@RestController
@RequestMapping("/")
public class IndexController {
    @Autowired
    public LocationService service;
    Logger logger = LoggerFactory.getLogger(IndexController.class);

    @GetMapping("/points")
    public List<Point> getPoints() {
        return service.getLocationList();
    }

    @PostMapping("/log")
    public void log(double lng, double lat) {
        logger.info("{},{}", lng, lat);
    }
}

常见问题：

Tomcat项目乱码

Tomcat中部署的JSP页面出现中文乱码问题：

1. 修改tomcat/conf目录下的主配置文件server.xml，添加URIEncoding=“UTF-8”配置项，配置位置如下：

   <Connector port="8080" protocol="HTTP/1.1"
          connectionTimeout="20000"
          redirectPort="8443" URIEncoding="UTF-8" />
   
   <Connector protocol="AJP/1.3"
          address="::1"
          port="8009"
          redirectPort="8443" URIEncoding="UTF-8" />

2. 修改tomcat/conf/web.xml，在 <servlet>节点中添加如下内容：

   <init-param>
          <param-name>fileEncoding</param-name>
          <param-value>UTF-8</param-value>
   </init-param>

3.重启服务

Tomcat设为开机自启

1. 进入init.d目录

进入到/etc/init.d目录下，命令是：

cd /etc/init.d

2. 新建一个名为tomcat的文件

vim tomcat

3. 为/etc/init.d/tomcat文件添加可执行权限

chmod 755 tomcat

4. 编辑tomcat文件,添加以下内容

vi tomcat

添加内容为： 注意CATALINA_HOME需要修改正确

#!/bin/bash
# processname: tomcat9
# chkconfig: 2345 86 16
# description: Tomcat9 start|restart|stop.

if [ -f /etc/init.d/functions ]; then
. /etc/init.d/functions
elif [ -f /etc/rc.d/init.d/functions ]; then
. /etc/rc.d/init.d/functions
else
echo -e "/atomcat: unable to locate functions lib. Cannot continue."
exit -1
fi

RETVAL=$?
CATALINA_HOME=/opt/pkg/tomcat9

case "$1" in
start)
if [ -f $CATALINA_HOME/bin/startup.sh ];
then
echo $"Starting Tomcat"
$CATALINA_HOME/bin/startup.sh
fi
;;
stop)
if [ -f $CATALINA_HOME/bin/shutdown.sh ];
then
echo $"Stopping Tomcat"
$CATALINA_HOME/bin/shutdown.sh
fi
;;
*)
echo $"Usage: $0 {start|stop}"
exit 1
;;
esac

exit $RETVAL

5. 把tomcat这个脚本添加到开机启动项里面

chkconfig --add tomcat
chkconfig tomcat on

6. 如果想看看是否添加成功

chkconfig --list

netconsole      0:关 1:关 2:关 3:关 4:关 5:关 6:关
network         0:关 1:关 2:开 3:开 4:开 5:开 6:关
tomcat          0:关 1:关 2:开 3:开 4:开 5:开 6:关

7. 在tomcat/bin下创建一个setenv.sh文件，加入以下环境变量, 并赋予执行权限

[root@hadoop000 tomcat9]# vi bin/setenv.sh

export JAVA_HOME=/opt/pkg/jdk1.8.0_261
export JRE_HOME=/opt/pkg/jdk1.8.0_261/jre
export CATALINA_HOME=/opt/pkg/tomcat9
export CATALINA_BASE=/opt/pkg/tomcat9

[root@hadoop000 tomcat9]# chmod a+x bin/setenv.sh

8. 查看看是否开机启动

使用命令重启机器，命令是：

reboot

9. 查看网络状态，执行命令,查看8080端口是否启动

netstat   -lntup

10. 查看tomcat进程

 ps -ef |grep tomcat

11. 如果不需要开机启动，从启动脚本删除即可

chkconfig --del tomcat

Views: 418

为什么使用Trident

逐个处理单个tuple会增加很多开销，因此storm中引入Trident实现batch处理.

Trident优点是：

• 批次处理消息
• 减少持久化的开销
• 结合Trident State能可靠保证每个消息只被处理一次

Trident的 State

Trident 在进行聚合操作时需要缓存中间结果, 可以看做Trident的状态(State).
Trident状态既可以保留在topology的内部，比如说内存中，也可以放到外部存储当中，比如说Memcached或者Cassandra数据库中.

Trident 允许以一种容错的方式来管理状态, 可以保证当遇到重试或错误时状态的更新是幂等的, 以此来实现EOS(Exactly only semantics)。

注：在数据统计分析中，幂等性是一个很重要的指标，因为它可以保证即使数据被处理了多次，但是站在结果的角度看和处理一次完全一样。

使用Trident实现词频统计

这里我们词频统计使用Trident拓扑实现.
这里使用Storm提供的MemoryMapState管理状态(每个单词的实时统计个数), 顾名思义MemoryMapState管理的状态是保存在内存中的. 状态可以通过方法stateQuery进行实时查询.

然后提供一个DRPC服务, 客户端可以指定需要查询词频的单词有哪些.

本地提交模式

数据源

在这个例子中，我们使用FixedBatchSpout 对象模拟数据源来发送一句一句的文本内容。输入数据源也可以和Kestrel或者Kafka这样的消息队列对接, Trident在处理输入流的时候会转换成若干个tuple组成的batch来处理。这个 FixedBatchSpout 的maxBatchSize为5.

FixedBatchSpout spout = new FixedBatchSpout(new Fields("sentence"), 5,
                new Values("the cow jumped over the moon"),
                new Values("the man went to the store and bought some candy"),
                new Values("four score and seven years ago"),
                new Values("how many apples can you eat"),
                new Values("to be or not to be the person"));
        spout.setCycle(false);

Trident实现词频统计

接下来我们创建TridentTopology计算词频, 并将实时统计的结果保存在TridentState中:

TridentTopology topology = new TridentTopology();

        Config conf = new Config();
        conf.setMaxSpoutPending(20);
        conf.setNumWorkers(3);

        LocalDRPC drpc = new LocalDRPC();
        LocalCluster localCluster = new LocalCluster();

        TridentState wordCounts =
                topology.newStream("spout1", spout).parallelismHint(16)
                        .each(new Fields("sentence"), new Split(), new Fields("word"))
                        .groupBy(new Fields("word"))
                        .persistentAggregate(new MemoryMapState.Factory(), new Count(), new Fields("count"))
                        .parallelismHint(3);

解析:

• TridentTopology的newStream方法传入了一个spout对象，spout对象会从外部读取数据并输出到当前topology当中，从而在topology中创建了一个新的数据流.
• Trident会在Zookeeper中保存一小部分状态信息来追踪数据的处理情况，而在代码中我们指定的字符串“spout1”就是Zookeeper中用来存储metadata信息的Znode节点.
• persistentAggregate方法会把数据流转换成一个TridentState对象, TridentState记录了单词的实时词频。

处理过程:

• 根据空格拆分sentence，并将拆分出的每个单词作为一个tuple输出
• 根据“word”字段进行groupBy操作
• persistentAggregate会帮助你把计数的结果进行存储

提供DRPC服务实现实时词频查询

通过newDRPCStream可以接受DRPC客户端请求的参数:

topology.newDRPCStream("words", drpc)
                .each(new Fields("args"), new Split(), new Fields("word"))
                .groupBy(new Fields("word"))
                .stateQuery(wordCounts, new Fields("word"), new MapGet(), new Fields("count"))
                .each(new Fields("count"), new FilterNull())
                .project(new Fields("word", "count"))
                .aggregate(new Fields("count"), new Sum(), new Fields("sum"));

解析:

• 对参数按照空格切分后作为word字段发射出去, 再对word 进行groupBy操作.
• 使用stateQuery来在上面代码中创建的TridentState对象上进行查询。
• stateQuery利用MapGet来获取每个单词的出现个数。
• 由于DRPC stream是使用跟TridentState完全同样的group方式（按照“word”字段进行group），每个单词的查询会被路由到TridentState的相应分区去执行。
• 用FilterNull这个过滤器把从未出现过的单词给去掉，
• 并使用Sum这个聚合器将这些词频统计结果累加起来。最终，Trident会自动把这个结果发送回等待的客户端。

提交到集群

为了更快看到结果, 这里使用本地提交方式:

        localCluster.submitTopology(topoName, conf, topology.build());

DRPC本地客户端发起实时查询

下面这部分实现了一个低延时的单词数量的分布式实时DRPC查询。这个查询以一个用空格分割的单词列表为输入，并返回这些单词事实出现次数。

这些查询是像普通的RPC调用那样被执行的，要说不同的话，那就是他们在后台是并行执行的。下面是执行DRPC实时查询的一个例子：

        for (int i = 0; i < 10; i++) {
            System.out.println("DRPC RESULT: " + drpc.execute("words", "cat the dog jumped"));
            Thread.sleep(1000);
        }

        localCluster.shutdown();
        drpc.shutdown();

解析:

• 发起DRPC请求, 请求的参数是"cat the dog jumped", 每隔1秒查询一次, 共10次.

Trident在如何最大程度的保证执行topogloy性能方面是非常智能的。在topology中会自动的发生两件非常有意思的事情：

• 更新状态和读取状态操作 (比如说 persistentAggregate 和 stateQuery) 会自动的是batch的形式操作状态。
• 如果有20次更新需要被同步到存储中，Trident会自动的把这些操作汇总到一起批处理，只做一次读写操作，而不是进行20次读写操作。

因此你可以在很方便的执行计算的同时，保证了非常好的性能。

Trident 的聚合器已经是被优化的非常好了的。Trident并不是简单的把一个group中所有的tuples都发送到同一个机器上面进行聚合，而是在发送之前已经进行过一次局部的聚合。打个比方，Count聚合器会先在每个partition上面进行count，然后把每个分片count汇总到一起就得到了最终 的count。这个技术其实就跟MapReduce里面的combiner是一个思想。

完整代码代码:

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.LocalDRPC;
import org.apache.storm.trident.TridentState;
import org.apache.storm.trident.TridentTopology;
import org.apache.storm.trident.operation.BaseFunction;
import org.apache.storm.trident.operation.TridentCollector;
import org.apache.storm.trident.operation.builtin.Count;
import org.apache.storm.trident.operation.builtin.FilterNull;
import org.apache.storm.trident.operation.builtin.MapGet;
import org.apache.storm.trident.testing.FixedBatchSpout;
import org.apache.storm.trident.testing.MemoryMapState;
import org.apache.storm.trident.tuple.TridentTuple;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

/**
 * 将wordcount结果持久化到数据库
 */
public class AxTridentWordCountLocally {

    public static void main(String[] args) throws Exception {
        String topoName = "wordCounter";

        FixedBatchSpout spout = new FixedBatchSpout(new Fields("sentence"), 5,
                new Values("the cow jumped over the moon"),
                new Values("the man went to the store and bought some candy"),
                new Values("four score and seven years ago"),
                new Values("how many apples can you eat"),
                new Values("to be or not to be the person"));
        spout.setCycle(false);

        TridentTopology topology = new TridentTopology();

        Config conf = new Config();
        conf.setMaxSpoutPending(20);
        conf.setNumWorkers(3);

        LocalDRPC localDRPC = new LocalDRPC();
        LocalCluster localCluster = new LocalCluster();

        /*
         * 根据数据源拆分单词后,然后分区操作,在每个分区上又进行分组(hash算法),然后在每个分组上进行聚合
         * 所以这里可能有多个分区,每个分区有多个分组,然后在多个分组上进行聚合
         * 用来进行group的字段会以key的形式存在于State当中，聚合后的结果会以value的形式存储在State当中
         */
        TridentState wordCounts = topology.newStream("spout1", spout).parallelismHint(16)
                .each(new Fields("sentence"), new Split(), new Fields("word"))
                .groupBy(new Fields("word"))
                .persistentAggregate(new MemoryMapState.Factory(), new Count(), new Fields("count"))
                .parallelismHint(3);

        topology.newDRPCStream("words", localDRPC).each(new Fields("args"), new Split(), new Fields("word"))
                .groupBy(new Fields("word"))
                .stateQuery(wordCounts, new Fields("word"), new MapGet(), new Fields("count"))
                .each(new Fields("count"), new FilterNull())
                .project(new Fields("word", "count"));

        localCluster.submitTopology(topoName, conf, topology.build()); //异步

        for (int i = 0; i < 10; i++) {
            System.out.println("DRPC RESULT: " + localDRPC.execute("words", "cat the dog jumped"));
            Thread.sleep(1000);
        }

        localCluster.shutdown();
        localDRPC.shutdown();
    }

    public static class Split extends BaseFunction {
        @Override
        public void execute(TridentTuple tuple, TridentCollector collector) {
            String sentence = tuple.getString(0);
            for (String word : sentence.split(" ")) {
                collector.emit(new Values(word));
            }
        }
    }
}

执行结果

在执行上述代码之后，可能输出如下所示：

DRPC RESULT: []
DRPC RESULT: [["the",30],["jumped",6]]
DRPC RESULT: [["the",84],["jumped",16]]
DRPC RESULT: [["jumped",26],["the",130]]
DRPC RESULT: [["jumped",30],["the",149]]
DRPC RESULT: [["jumped",40],["the",199]]
DRPC RESULT: [["the",245],["jumped",49]]
DRPC RESULT: [["jumped",59],["the",295]]
DRPC RESULT: [["the",345],["jumped",69]]
DRPC RESULT: [["the",394],["jumped",79]]

数据流向示意图

远程提交模式

import org.apache.storm.Config;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.trident.TridentState;
import org.apache.storm.trident.TridentTopology;
import org.apache.storm.trident.operation.BaseFunction;
import org.apache.storm.trident.operation.TridentCollector;
import org.apache.storm.trident.operation.builtin.Count;
import org.apache.storm.trident.operation.builtin.FilterNull;
import org.apache.storm.trident.operation.builtin.MapGet;
import org.apache.storm.trident.testing.FixedBatchSpout;
import org.apache.storm.trident.testing.MemoryMapState;
import org.apache.storm.trident.tuple.TridentTuple;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import org.apache.storm.utils.DRPCClient;

public class CxTridentWordCountRemotely {
    public static StormTopology buildTopology() {
        FixedBatchSpout spout = new FixedBatchSpout(new Fields("sentence"), 3, new Values("the cow jumped over the moon"),
                new Values("the man went to the store and bought some candy"),
                new Values("four score and seven years ago"),
                new Values("how many apples can you eat"), new Values("to be or not to be the person"));
        spout.setCycle(true);

        TridentTopology topology = new TridentTopology();

        TridentState wordCounts = topology.newStream("spout1", spout).parallelismHint(16).each(new Fields("sentence"),
                new Split(), new Fields("word"))
                .groupBy(new Fields("word")).persistentAggregate(new MemoryMapState.Factory(),
                        new Count(), new Fields("count"))
                .parallelismHint(16);

        topology.newDRPCStream("words").each(new Fields("args"), new Split(), new Fields("word"))
                .groupBy(new Fields("word"))
                .stateQuery(wordCounts, new Fields("word"), new MapGet(), new Fields("count"))
                .each(new Fields("count"), new FilterNull())
                .project(new Fields("word", "count"));
        return topology.build();
    }

    public static void main(String[] args) throws Exception {
        Config conf = new Config();
        conf.setMaxSpoutPending(20);
        String topoName = "wordCounter";
        if (args.length > 0) {
            topoName = args[0];
        }
        conf.setNumWorkers(3);
        StormSubmitter.submitTopologyWithProgressBar(topoName, conf, buildTopology());
        try (DRPCClient drpc = DRPCClient.getConfiguredClient(conf)) {
            for (int i = 0; i < 10; i++) {
                System.out.println("DRPC RESULT: " + drpc.execute("words", "cat the dog jumped"));
                Thread.sleep(1000);
            }
        }
    }
    public static class Split extends BaseFunction {
        @Override
        public void execute(TridentTuple tuple, TridentCollector collector) {
            String sentence = tuple.getString(0);
            for (String word : sentence.split(" ")) {
                collector.emit(new Values(word));
            }
        }
    }
}

打成jar包上传到集群

Views: 627

需求

针对twitter网站上的一篇推文的接触用户(也叫REACH值）进行统计。

Reach值让你了解推文的真实覆盖到的用户群体, 要计算一个推文URL的Reach值，需要以下4步：

1. 根据推文的URL查询数据库获取全部直接接触用户（转发的用户）
2. 再根据接触用户通过查询数据库获取每个用户的全部粉丝
3. 对粉丝集合中的用户进行去重处理
4. 最后统计去重后的用户数, 即这个推文的Reach值

拓扑定义

一个单独的Reach计算在计算期间可能涉及到数千次数据库访问和数千万的粉丝记录查询，可能是一个非常耗时的计算。在storm上实现这个功能非常简单。在一台机器上，Reach计算可能花费数分钟，而在storm集群，最难计算Reach的URL也只需数秒。

Storm-starter 项目中有一个计算Reach样例，Reach拓扑定义如下所示：

LinearDRPCTopologyBuilder builder = new LinearDRPCTopologyBuilder("reach"); 

builder.addBolt(new GetTweeters(), 3); 
builder.addBolt(new GetFollowers(), 2) .shuffleGrouping(); 
builder.addBolt(new PartialUniquer(), 3) .fieldsGrouping(new Fields("id", "follower")); 
builder.addBolt(new CountAggregator()) .fieldsGrouping(new Fields("id")); 

过程分析

以计算https://tech.backtype.com/blog/123的Reach值为例进行说明:

这个拓扑以4个步骤的形式执行, 因此一共有4个Bolt：

1. GetTweeters 从数据库获取给定URL对应的用户并发射出去
2. GetFollowers 从数据库获取每个用户对应的粉丝并发射出去
3. PartialUniquer按粉丝进行字段分组并利用Set集合进行去重处理, 将去重后的粉丝数量进行局部累加, 并把结果发送出去
4. 最后，CountAggregator从每个PartialUniquer任务接收计数并对累加求和作为返回值给DRPC客户端。

项目代码

为了方便项目中数据库使用Map集合进行伪造, 代码如下:

package storm.example.drpc;

/**
 * 计算推文的REACH值
 * storm.example.drpc.AdvanceDRPCTopology
 */

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.LocalDRPC;
import org.apache.storm.StormSubmitter;
import org.apache.storm.coordination.BatchOutputCollector;
import org.apache.storm.drpc.LinearDRPCTopologyBuilder;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.topology.base.BaseBatchBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import org.apache.storm.utils.Utils;

import java.util.*;

// 计算一篇推文的REACH值
public class AdvanceDRPCTopology {

    // MAP: URL -> USER
    public static Map<String, List<String>> TWEETERS_DB = new HashMap<String, List<String>>() {{
        put("foo.com/blog/1", Arrays.asList("sally", "bob", "tim", "george", "nathan"));
        put("engineering.twitter.com/blog/5", Arrays.asList("adam", "david", "sally", "nathan"));
        put("tech.backtype.com/blog/123", Arrays.asList("tim", "mike", "john"));
    }};

    // MAP: USER -> FOLLOWERS
    public static Map<String, List<String>> FOLLOWERS_DB = new HashMap<String, List<String>>() {{
        put("sally", Arrays.asList("bob", "tim", "alice", "adam", "jim", "chris", "jai"));
        put("bob", Arrays.asList("sally", "nathan", "jim", "mary", "david", "vivian"));
        put("tim", Arrays.asList("alex"));
        put("nathan", Arrays.asList("sally", "bob", "adam", "harry", "chris", "vivian", "emily", "jordan"));
        put("adam", Arrays.asList("david", "carissa"));
        put("mike", Arrays.asList("john", "bob"));
        put("john", Arrays.asList("alice", "nathan", "jim", "mike", "bob"));
    }};

    public static LinearDRPCTopologyBuilder construct() {

        LinearDRPCTopologyBuilder builder = new LinearDRPCTopologyBuilder("reach");

        builder.addBolt(new GetTweeters(), 3);
        builder.addBolt(new GetFollowers(), 2).shuffleGrouping();
        builder.addBolt(new PartialUniquer(), 3).fieldsGrouping(new Fields("id", "follower"));
        builder.addBolt(new CountAggregator(), 1).fieldsGrouping(new Fields("id"));

        return builder;
    }

    public static void main(String[] args) throws Exception {

        LinearDRPCTopologyBuilder builder = construct();
        Config conf = new Config();

        if (args == null || args.length == 0) {
            // conf.setMaxTaskParallelism(3);
            LocalDRPC drpc = new LocalDRPC();
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("reach-drpc", conf, builder.createLocalTopology(drpc));

            String[] urlsToTry = new String[]{"foo.com/blog/1", "engineering.twitter.com/blog/5", "notaurl.com", "tech.backtype.com/blog/123"};

            for (String url : urlsToTry) {
                System.err.println("Thread[" + Thread.currentThread().getName() + "] " +"Reach of " + url + ": " + drpc.execute("reach", url));
                Utils.sleep(10000);
            }
            cluster.shutdown();
            drpc.shutdown();
        } else {
            conf.setNumWorkers(1); //6
            StormSubmitter.submitTopology(args[0], conf, builder.createRemoteTopology());
        }
    }

    // BOLT 1: 根据URL发送推特用户到数据流中
    public static class GetTweeters extends BaseBasicBolt {

        @Override
        public void execute(Tuple tuple, BasicOutputCollector collector) {

            Object id = tuple.getValue(0);
            String url = tuple.getString(1);
            List<String> tweeters = TWEETERS_DB.get(url);

            if (tweeters != null) {
                for (String tweeter : tweeters) {
                    collector.emit(new Values(id, tweeter));
                }
            }
        }

        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {

            declarer.declare(new Fields("id", "tweeter"));
        }

    }

    // BOLT 2: 根据推特用户发送对应的粉丝到数据流中
    public static class GetFollowers extends BaseBasicBolt {

        @Override
        public void execute(Tuple tuple, BasicOutputCollector collector) {

            Object id = tuple.getValue(0);
            String tweeter = tuple.getString(1);
            List<String> followers = FOLLOWERS_DB.get(tweeter);

            if (followers != null) {
                for (String follower : followers) {
                    System.err.println("Thread[" + Thread.currentThread().getName() + "] " + "request-id[" + id + "]: " + tweeter + "'s follower: " + follower);
                    collector.emit(new Values(id, follower));
                }
            }
        }

        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {

            declarer.declare(new Fields("id", "follower"));
        }
    }

    // BOLT 3: 对粉丝去重后的计数发送的数据流中
    public static class PartialUniquer extends BaseBatchBolt {

        BatchOutputCollector collector;
        Object id;
        Set<String> followers = new HashSet<String>();

        @Override
        public void prepare(Map conf, TopologyContext context, BatchOutputCollector collector, Object id) {
            this.collector = collector;
            this.id = id;
        }

        @Override
        public void execute(Tuple tuple) {
            followers.add(tuple.getString(1));
        }

        @Override
        public void finishBatch() {
            System.err.println("Thread[" + Thread.currentThread().getName() + "] " +"request-id[" + id + "]: " + "followers.size(): " + followers.size() + followers);
            collector.emit(new Values(id, followers.size()));
        }

        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            declarer.declare(new Fields("id", "partial-count"));
        }
    }

    // BOLT 4：将所有计数累加的结果发送的数据流中
    public static class CountAggregator extends BaseBatchBolt {

        BatchOutputCollector collector;

        Object id;
        int reach = 0;

        @Override
        public void prepare(Map conf, TopologyContext context, BatchOutputCollector collector, Object id) {
            this.collector = collector;
            this.id = id;
        }

        @Override
        public void execute(Tuple tuple) {
            int partiaLReach = tuple.getInteger(1);
            System.err.println("Thread[" + Thread.currentThread().getName() + "] " +"request-id[" + id + "]: " + "Partial reach is " + partiaLReach);
            reach += tuple.getInteger(1);
        }

        @Override
        public void finishBatch() {
            System.err.println("Thread[" + Thread.currentThread().getName() + "] " +"request-id[" + id + "]: " + "Global reach is " + reach);
            collector.emit(new Values(id, reach));
        }

        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            declarer.declare(new Fields("id", "reach"));
        }
    }
}

Reach计算的每一单步都是可以并行执行的，而且定义一个DRPC拓扑也非常简单。

需要注意的是, CountAggregator 只有在 PartialUniquer 并行度大于1个时候才有意义.

另外CountAggregator 和 PartialUniquer 这两个Bolt都是继承的BaseBatchBolt,这种Bolt发射数据流的时机是在相同id的数据都处理完成后, 才会触发(类似事务).

谢谢!

Views: 355

分布式远程过程调用

所谓过程调用就是指方法调用, 讲解分布式远程过程调用前先看一看什么是普通方法调用和远程过程调用.

1. 普通方法调用

以一个普通Java类中的方法为例:

// 方法定义
public String exclamation(String arg) {
    return arg + "!";
}
// 方法调用
instance.exclamation("hi"); // hi!

2. 远程过程调用

RPC = Remote Procedure Call(远程过程调用)

RPC即在一台机器以远程的方式调用另外一台机器中的应用中的功能(方法或函数)

Storm中使用的Thrift就是一个RPC框架, Thrift由C++编写，但可以支持很多语言。要创建一个Thrift服务，必须写一些Thrift文件来描述它，为目标语言生成代码，并且写一些代码来启动服务器及从客户端调用它。其实Topology 的定义就是一个 Thrift 结构，并且 Nimbus 就是一个 Thrift 服务，你可以提交任务语言创建的Topology到Nimbus。

3. 分布式远程过程调用

DRPC, 即分布式RPC（distributed RPC，DRPC）,即远程调用的结果是通过集群中的并行计算返回的.

Storm中的DRPC只是通过以下基本元素streams，spout，bolt，topology衍生的一个模式, 算不上Storm的主要特性。Storm的DRPC可以单独作为一个独立于Storm的库发布，但由于其重要性还是和Storm捆绑在了一起。

Storm的DRPC就是每一次函数调用都是通过DRPC客户端以请求的形式提交给集群中的DRPC服务器, 然后在提交的DRPC拓扑中寻找到对应的函数在集群中进行并行计算, DRPC拓扑接收调用函数的参数信息作为输入流，并将计算结果作为输出流发射出去,最终返回给DRPC客户端。

总体概述

DRPC通过DRPC Server来实现，DRPC Server的整体工作过程如下：

1. 接收到一个RPC调用请求
2. 发送请求到Storm上的拓扑
3. 从Storm上接收计算结果
4. 将计算结果返回给客户端

1. Client向DRPC Server发送被调用执行的DRPC函数名称及参数
2. Storm上的topology通过DRPCSpout实现这一函数，从DPRC Server接收到函数调用流
3. DRPC Server会为每次函数调用生成唯一的id
4. Storm上运行的topology开始计算结果，最后通过一个ReturnResults的Bolt连接到DRPC Server，发送指定id的计算结果
5. DRPC Server通过使用之前为每个函数调用生成的id，将结果关联到对应的发起调用的client，将计算结果返回给client

LinearDRPCTopologyBuilder

Storm提供了一个topology builder——LinearDRPCTopologyBuilder，它可以自动完成几乎所有的DRPC步骤。包括：

1. 构建spout；
2. 向DRPC Server返回结果；
3. 为Bolt提供函数用于对tuples进行聚集。

下面是一个简单的例子，这个DRPC拓扑只是简单的在输入参数后追加"!"后返回：

public static class ExclaimBolt extends BaseBasicBolt {
    public void execute(Tuple tuple, BasicOutputCollector collector) {
        String input = tuple.getString(1);
        collector.emit(new Values(tuple.getValue(0), input + "!"));
    }

    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("id", "result"));
    }
}

public static void main(String[] args) throws Exception {
    LinearDRPCTopologyBuilder builder = new LinearDRPCTopologyBuilder("exclamation");
    builder.addBolt(new ExclaimBolt(), 3);
    // ...
}

由上述例子可见，我们只需很少的工作即可完成拓扑。当创建LinearDRPCTopologyBuilder的时候，需要指定拓扑中DRPC函数的名称exclamation。一个DRPC Server可以协调多个函数，每个函数有不同的函数名称。拓扑中的第一个bolt的输入是两个字段：第一个是请求的id；第二个是请求的参数。

LinearDRPCTopologyBuilder同时需要最后一个bolt发射一个包含两个字段的输出流：第一个字段是请求id；第二个字段是计算结果。因此，所有的中间tuple必须包含请求<code>id作为第一个字段。

例子中，ExclaimBolt在输入tuple的第二个字段后面追加"!"，LinearDRPCTopologyBuilder负责处理其余的协调工作：与DRPC Server建立连接，发送结果给DRPC Server。

Local Mode DRPC

DRPC可以以本地模式运行，下面的代码是如何在本地模式运行上面的例子：

package storm.example.drpc;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.LocalDRPC;
import org.apache.storm.drpc.LinearDRPCTopologyBuilder;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

/**
 * @Author: deLucia
 * @Version: 1.0
 * @Description: Locally DRPC
 * Program arguments, eg.: drpc drpc-fun 1 2 3 one two three
 */
public class BasicDRPCTopologyLocally {

    // id : request id
    // result: request parameter
    public static class ExclaimBolt extends BaseBasicBolt {
        @Override
        public void execute(Tuple tuple, BasicOutputCollector collector) {
            Object id = tuple.getValue(0);
            String input = tuple.getString(1);
            collector.emit(new Values(id, input + "!"));
        }

        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            declarer.declare(new Fields("id", "result"));
        }
    }

    public static void main(String[] args) throws Exception {
        Config conf = new Config();
        String topoName = "DRPCExample";
        String function = "exclamation";
        if (args != null) {
            if (args.length > 0) {
                topoName = args[0];
            }
            if (args.length > 1) {
                function = args[1];
            }
        }

        LinearDRPCTopologyBuilder builder = new LinearDRPCTopologyBuilder(function);
        builder.addBolt(new ExclaimBolt(), 2);

        conf.setNumWorkers(1);

        // LOCAL DRPC
        LocalDRPC localDRPC = new LocalDRPC();
        LocalCluster localCluster = new LocalCluster();
        localCluster.submitTopology(topoName, conf, builder.createLocalTopology(localDRPC));

        if (args != null && args.length > 2) {

            // DRPC Client
            for (int i = 2; i < args.length; i++) {
                String word = args[i];
                System.out.println("Result for \"" + word + "\": " + localDRPC.execute(function, word));
            }
        }

        localCluster.shutdown();
        localDRPC.shutdown();
    }
}

Remote Mode DRPC

在实际的Storm集群上运行DRPC也一样很简单。只需完成以下步骤：

1. 配置DRPC Server(s)地址:

   drpc.servers:
     - "hadoop000"
   drpc.http.port: 8081
   #storm.thrift.transport: "org.apache.storm.security.auth.plain.PlainSaslTransportPlugin"
   
   nimbus.thrift.port: 16627

   注意nimbus.thrift.port的默认端口为6627, 这里改成了16627

2. 启动DRPC Server(s)

   $ bin/storm drpc 

3. 向Storm集群提交DRPC拓扑。

   $ bin/storm jar ...

首先，通过storm脚本启动DRPC Server：

然后，在Storm集群中配置DRPC Server地址，这就是DRPCSpout读取函数调用请求的地方。这一步的配置可以通过storm.yaml文件或者拓扑的配置来完成。通过storm.yaml文件的配置方式如下：`

drpc.servers:
  - "drpc1.foo.com"
  - "drpc2.foo.com"

最后，通过StormSubmitter启动<code>DRPC拓扑。为了以远程模式运行上面的例子，代码如下：

StormSubmitter.submitTopology("exclamation-drpc", conf, builder.createRemoteTopology());

createRemoteTopology被用于为Storm集群创建合适的拓扑。

package storm.example.drpc;

import org.apache.storm.Config;
import org.apache.storm.StormSubmitter;
import org.apache.storm.drpc.LinearDRPCTopologyBuilder;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import org.apache.storm.utils.DRPCClient;

/**
 * @Author: deLucia
 * @Version: 1.0
 * @Description: Upload to Storm Cluster
 */
public class BasicDRPCTopologyRemotely {
    public static void main(String[] args) throws Exception {
        Config conf = new Config();
        String topoName = "DRPCExample";
        String function = "exclamation";
        if (args != null) {
            if (args.length > 0) {
                topoName = args[0];
            }
            if (args.length > 1) {
                function = args[1];
            }
        }

        LinearDRPCTopologyBuilder builder = new LinearDRPCTopologyBuilder(function);
        builder.addBolt(new ExclaimBolt(), 2);

        conf.setNumWorkers(1);

        // REMOTE DPRC
        StormSubmitter.submitTopologyWithProgressBar(topoName, conf, builder.createRemoteTopology());

        if (args != null && args.length > 2) {

             // REMOTE DRPC
             try (DRPCClient drpc = DRPCClient.getConfiguredClient(conf)) {
                 for (int i = 2; i < args.length; i++) {
                     String word = args[i];
                     System.out.println("Result for \"" + word + "\": " + drpc.execute(function, word));
                 }
             }
        }
    }

    public static class ExclaimBolt extends BaseBasicBolt {
        @Override
        public void execute(Tuple tuple, BasicOutputCollector collector) {
            String input = tuple.getString(1);
            collector.emit(new Values(tuple.getValue(0), input + "!"));
        }

        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            declarer.declare(new Fields("id", "result"));
        }
    }
}

发起DRPC请求的三种方式

要向DRPC服务器提交请求有三种方式:

1. 服务器上通过 shell 命令调用
   语法: bin/storm drpc-client function-name function-args
2. 通过http的方式发起Get请求 http://hostname:drpc-http-port/drpc/function-name/function-args, 如果是在浏览器的地址栏输入则结果将直接显示在页面上(注意一些特殊字符需要urlencode)
3. 客户端程序远程调用(3772为drpc server默认的端口号):
   DRPCClient client = new DRPCClient("hostname", 3772);

Views: 396