Trident中的DRPC实现

一：介绍

1.说明

　　Storm里面引入DRPC主要是利用storm的实时计算能力来并行化CPU intensive的计算。DRPC的storm topology以函数的参数流作为输入，而把这些函数调用的返回值作为topology的输出流。

2.工作机制

　　Distributed RPC是由一个”DPRC Server”协调的(storm自带了一个实现)DRPC服务器协调

　　1) 接收一个RPC请求。

　　2) 发送请求到storm topology 

　　3) 从storm topology接收结果。

　　4) 把结果发回给等待的客户端。从客户端的角度来看一个DRPC调用跟一个普通的RPC调用没有任何区别。

3.工作流程

　　

　　客户端给DRPC服务器发送要执行的方法的名字，以及这个方法的参数。

　　实现了这个函数的topology使用 DRPCSpout 从DRPC服务器接收函数调用流。

　　每个函数调用被DRPC服务器标记了一个唯一的id。 这个topology然后计算结果，在topology的最后一个叫做 ReturnResults 的bolt会连接到DRPC服务器，并且把这个调用的结果发送给DRPC服务器(通过那个唯一的id标识)。DRPC服务器用那个唯一id来跟等待的客户端匹配上，唤醒这个客户端并且把结果发送给它。

二：本地DRPC

1.主驱动类

package com.jun.tridentWithHbase;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.LocalDRPC;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import org.apache.storm.hbase.trident.state.HBaseMapState;
import storm.trident.Stream;
import storm.trident.TridentState;
import storm.trident.TridentTopology;
import storm.trident.operation.builtin.Count;
import storm.trident.operation.builtin.MapGet;
import storm.trident.operation.builtin.Sum;
import storm.trident.state.OpaqueValue;
import storm.trident.state.StateFactory;
import storm.trident.testing.FixedBatchSpout;
import storm.trident.testing.MemoryMapState;

public class TridentDemo {
    public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException {
        TridentTopology tridentTopology=new TridentTopology();
        //模拟数据
        Fields field=new Fields("log","flag");
        FixedBatchSpout spout=new FixedBatchSpout(field,5,
            new Values("168.214.187.214 - - [1481953616092] "GET /view.php HTTP/1.1" 200 0 "http://cn.bing.com/search?q=spark mllib" "Mozilla/5.0 (Windows NT 6.1; rv:2.0.1) Gecko/20100101 Firefox/4.0.1" "-"","A"),
            new Values("168.187.202.202 - - [1481953537038] "GET /IBEIfeng.gif?order_id=1063&orderTime=1481953537038&memberId=4000012340500607&productInfos=10005-2099.48-B-1|10004-1886.62-A-2|10001-961.99-A-1&orderAmt=6834.70 HTTP/1.1" 200 0 "-" "Mozilla/5.0 (compatible; MSIE 10.0; Windows NT 6.2;Tident/6.0)" "-"","A"),
            new Values("61.30.167.187 - - [1481953539039] "GET /IBEIfeng.gif?order_id=1064&orderTime=1481953539039&memberId=4000930409959999&productInfos=10007-3329.13-B-1|10009-2607.71-B-1|10002-390.62-A-1|10006-411.00-B-2&orderAmt=7149.46 HTTP/1.1" 200 0 "-" "Mozilla/5.0 (Linux; Android 4.2.1; Galaxy Nexus Build/JOP40D) AppleWebKit/535.19 (KHTML, like Gecko) Chrome/18.0.1025.166 Mobile Safari/535.19" "-"","A"),
            new Values("30.29.132.190 - - [1481953544042] "GET /IBEIfeng.gif?order_id=1065&orderTime=1481953544043&memberId=1234568970080798&productInfos=10005-2099.48-B-1|10001-3242.40-C-2|10006-411.00-B-1&orderAmt=8995.28 HTTP/1.1" 200 0 "-" "Mozilla/5.0 (iPhone; CPU iPhone OS 7_)_3 like Mac OS X) AppleWebKit/537.51.1 (KHTML, like Gecko) Version/7.0 Mobile/11B511 Safari/9537.53" "-"","B"),
            new Values("222.190.187.201 - - [1481953578068] "GET /IBEIfeng.gif?order_id=1066&orderTime=1481953578068&memberId=3488586887970809&productInfos=10005-2099.48-B-1|10001-2774.16-C-2&orderAmt=7647.80 HTTP/1.1" 200 0 "-" "Mozilla/5.0 (Windows NT 6.1; rv:2.0.1) Gecko/20100101 Firefox/4.0.1" "-"","B"),
            new Values("72.202.43.53 - - [1481953579069] "GET /IBEIfeng.gif?order_id=1067&orderTime=1481953579069&memberId=2084859896989877&productInfos=10007-3329.13-B-1|10001-961.99-A-2&orderAmt=5253.10 HTTP/1.1" 200 0 "-" "Mozilla/5.0 (Linux; Android 4.2.1; Galaxy Nexus Build/JOP40D) AppleWebKit/535.19 (KHTML, like Gecko) Chrome/18.0.1025.166 Mobile Safari/535.19" "-"","B")
        );
        //多次循环
        spout.setCycle(true);
        //流处理
        Stream stream=tridentTopology.newStream("orderAnalyse",spout)
                //过滤
            .each(new Fields("log"),new ValidLogFilter())
                //解析
            .each(new Fields("log"), new LogParserFunction(),new Fields("orderId","orderTime","orderAmtStr","memberId"))
                //投影
            .project(new Fields("orderId","orderTime","orderAmtStr","memberId"))
                //时间解析
            .each(new Fields("orderTime"),new DateTransFormerFunction(),new Fields("day","hour","minter"))
         ;
        //分流
        //1.基于minter统计订单数量，分组统计
        TridentState state=stream.groupBy(new Fields("minter"))
                //全局聚合，使用内存存储状态信息
                .persistentAggregate(new MemoryMapState.Factory(),new Count(),new Fields("orderNumByMinter"));
//        state.newValuesStream().each(new Fields("minter","orderNumByMinter"),new PrintFilter());

        //2.另一个流,基于分钟的订单金额，局部聚合
        Stream partitionStream=stream.each(new Fields("orderAmtStr"),new TransforAmtToDoubleFunction(),new Fields("orderAmt"))
            .groupBy(new Fields("minter"))
                    //局部聚合
                .chainedAgg()    //聚合链
            .partitionAggregate(new Fields("orderAmt"),new LocalSum(),new Fields("orderAmtSumOfLocal"))
                .chainEnd();      //聚合链

        //做一次全局聚合
        TridentState partitionState=partitionStream.groupBy(new Fields("minter"))
                //全局聚合
                .persistentAggregate(new MemoryMapState.Factory(),new Fields("orderAmtSumOfLocal"),new Sum(),new Fields("totalOrderAmt"));
        partitionState.newValuesStream().each(new Fields("minter","totalOrderAmt"),new PrintFilter());

        //提交
        Config config=new Config();
        if(args==null || args.length<=0){
            //应该是构建一个DRPC的服务器
            LocalDRPC localDRPC=new LocalDRPC();
            tridentTopology.newDRPCStream("orderDataServer",localDRPC)
                    //参数处理
                .each(new Fields("args"),new RequestParamsParserFunction(),new Fields("date"))
                    //查询，重要的参数是上面的partitionState
                .stateQuery(partitionState,new Fields("date"),new MapGet(),new Fields("totalAmtByMinter"))
                    //投影
                .project(new Fields("date","totalAmtByMinter"));
            //提交任务
            LocalCluster localCluster=new LocalCluster();
            localCluster.submitTopology("tridentDemo",config,tridentTopology.build());
            //获取值
            String jsonResult=localDRPC.execute("orderDataServer","201612171345 201612171345");
            System.out.println("***"+jsonResult+"***");

        }else {
            config.setNumWorkers(2);
            StormSubmitter.submitTopology(args[0],config,tridentTopology.build());
        }
    }
}

2.请求参数处理类

package com.jun.tridentWithHbase;

import backtype.storm.tuple.Values;
import storm.trident.operation.Function;
import storm.trident.operation.TridentCollector;
import storm.trident.operation.TridentOperationContext;
import storm.trident.tuple.TridentTuple;

import java.util.Map;

public class RequestParamsParserFunction implements Function {
    @Override
    public void execute(TridentTuple tridentTuple, TridentCollector tridentCollector) {
        String parameters=tridentTuple.getStringByField("args");
        String[] params=parameters.split(" ");
        for (String param:params){
            tridentCollector.emit(new Values(param));
        }
    }

    @Override
    public void prepare(Map map, TridentOperationContext tridentOperationContext) {

    }

    @Override
    public void cleanup() {

    }
}

3.效果

　　

三：集群模式的DRPC　

1.主驱动类

 config.setNumWorkers(2);
            //集群上构建DRPC服务器
            tridentTopology.newDRPCStream("orderDataServer")
                    //参数处理
                    .each(new Fields("args"),new RequestParamsParserFunction(),new Fields("date"))
                    //查询，重要的参数是上面的partitionState
                    .stateQuery(partitionState,new Fields("date"),new MapGet(),new Fields("totalAmtByMinter"))
                    //投影
                    .project(new Fields("date","totalAmtByMinter"));
            StormSubmitter.submitTopology(args[0],config,tridentTopology.build());

2.配置DRPC服务和端口

　　

3.启动storm

4.启动Drpc进程

　　在drpc.servers参数所指定的服务器上。

　　命令：nohup bin/storm drpc >>/dev/null 2>&1 &

　　查看端口是否正常打开：netstat -tlnup | grep 3772

5.将jar包提交到集群上

6.编写客户端

package com.jun.tridentWithKafka;

import backtype.storm.generated.DRPCExecutionException;
import backtype.storm.utils.DRPCClient;
import org.apache.thrift7.TException;

public class DrpcClientDemo {
    public static void main(String[] args) {
        DRPCClient drpcClient=new DRPCClient("linux-hadoop01.ibeifeng.com",3772);
        try {
            String jsonResult=drpcClient.execute("orderDataServer","201612171345 201612171345");
            System.out.println("==="+jsonResult+"===");
        } catch (TException e) {
            e.printStackTrace();
        } catch (DRPCExecutionException e) {
            e.printStackTrace();
        }
    }
}