Storm流之FieldGrouping字段分组

心已赠人 2022-02-21 09:52 231阅读 0赞

一、      需求

字符串按照是否包含o分组

二、      Topology

package com.test.csdn;  
import com.test.storm.bolt.FieldPrintBolt;  
import com.test.storm.spout.SplitTestSpout;  
import org.apache.storm.Config;  
import org.apache.storm.LocalCluster;  
import org.apache.storm.topology.TopologyBuilder;  
import org.apache.storm.tuple.Fields;  
import org.apache.storm.utils.Utils;  
   
/\*\*  
 \* Created by Simon on 2018/4/26.  
 \*/  
public class FiledGroupingTopo \{  
    public static void main(String\[\] args) \{  
   
        TopologyBuilder builder = new TopologyBuilder();  
   
        builder.setSpout("spout", new FiledGroupingSpout(), 1);  
        //指定5个，便于测试  
        builder.setBolt("bolt", new FiledGroupingBolt(), 5).fieldsGrouping("spout", new Fields("isContainsO"));  
   
       Config conf = new Config();  
        conf.setDebug(false);  
   
        LocalCluster cluster = new LocalCluster();  
        cluster.submitTopology("toplogy", conf, builder.createTopology());  
        Utils.sleep(Long.MAX\_VALUE);  
        cluster.shutdown();  
   
    \}  
\}

三、      Spout

package com.test.csdn;  
import org.apache.storm.spout.SpoutOutputCollector;  
import org.apache.storm.task.TopologyContext;  
import org.apache.storm.topology.OutputFieldsDeclarer;  
import org.apache.storm.topology.base.BaseRichSpout;  
import org.apache.storm.tuple.Fields;  
import org.apache.storm.tuple.Values;  
   
   
import java.util.Map;  
import java.util.Random;  
   
/\*\*  
 \* Created by Simon on 2018/4/26.  
 \*/  
public class FiledGroupingSpout extends BaseRichSpout\{  
    private SpoutOutputCollector collector;  
    String\[\] str =  \{"xiaomi","huawei","apple","oppo","vivo","lenovo","LG",  
            "samsung","htc","honor","nokia","smartisan","Sony","BlackBerry","sharp"\};  
   
    @Override  
    public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) \{  
        this.collector=spoutOutputCollector;  
    \}  
   
    @Override  
    public void nextTuple() \{  
        try \{  
            Thread.sleep(1000);  
            int i = new Random().nextInt(15);  
            String string = str\[i\];  
            //按照是否有o分组  
            boolean o = string.contains("o");  
            collector.emit(new Values(o,string));  
        \} catch (InterruptedException e) \{  
            e.printStackTrace();  
        \}  
   
    \}  
   
    @Override  
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) \{  
        outputFieldsDeclarer.declare(new Fields("isContainsO","string"));  
    \}  
\}

四、      Bolt

package com.test.csdn;  
   
import org.apache.storm.topology.BasicOutputCollector;  
import org.apache.storm.topology.OutputFieldsDeclarer;  
import org.apache.storm.topology.base.BaseBasicBolt;  
import org.apache.storm.tuple.Tuple;  
   
/\*\*  
 \* Created by Simon on 2018/4/26.  
 \*/  
public class FiledGroupingBolt extends BaseBasicBolt\{  
   
    @Override  
    public void execute(Tuple tuple, BasicOutputCollector basicOutputCollector) \{  
        System.out.println(Thread.currentThread().getName()+"\_\_\_"+tuple.getValue(1));  
    \}  
   
    @Override  
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) \{  
   
    \}  
\}

五、      输出结果

![20180426120440787][]   
\---------------------   
作者：\_\_Simon   
来源：CSDN   
原文：https://blog.csdn.net/Simon\_09010817/article/details/80092080   
版权声明：本文为博主原创文章，转载请附上博文链接！

[20180426120440787]: /images/20220221/2b638e086bb140f4b439ed0cb354f602.png