1. 代码100分首页
  2. 代码基础

基于Storm的WordCount「建议收藏」

小小码 • • 代码基础 • 阅读 131
一键激活最新全家桶

基于Storm的WordCount「建议收藏」Storm WordCount 工作过程 Storm 版本： 1、Spout 从外部数据源中读取数据，随机发送一个元组对象出去； 2、SplitBolt 接收 Spout 中输出的元组对象，将元组中的

Storm WordCount 工作过程

Storm 版本：
1、Spout 从外部数据源中读取数据，随机发送一个元组对象出去；
2、SplitBolt 接收 Spout 中输出的元组对象，将元组中的数据切分成单词，并将切分后的单词发射出去；
3、WordCountBolt 接收 SplitBolt 中输出的单词数组，对里面单词的频率进行累加，将累加后的结果输出。

Java 版本：
1、读取文件中的数据，一行一行的读取；
2、将读到的数据进行切割；
3、对切割后的数组中的单词进行计算。

Hadoop 版本：
1、按行读取文件中的数据；
2、在 Mapper()函数中对每一行的数据进行切割，并输出切割后的数据数组；
3、接收 Mapper()中输出的数据数组，在 Reducer()函数中对数组中的单词进行计算，将计算后的统计结果输出。

源代码

storm的配置、eclipse里maven的配置以及创建项目部分省略。

Mainclass

package com.test.stormwordcount;
import backtype.storm.Config; 
import backtype.storm.LocalCluster; 
import backtype.storm.StormSubmitter; 
import backtype.storm.generated.AlreadyAliveException; 
import backtype.storm.generated.InvalidTopologyException; 
import backtype.storm.topology.TopologyBuilder; 
import backtype.storm.tuple.Fields; 

public class MainClass { 

    public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException {         
        //创建一个 TopologyBuilder         
        TopologyBuilder tb = new TopologyBuilder();         
        tb.setSpout("SpoutBolt", new SpoutBolt(), 2);         tb.setBolt("SplitBolt", new SplitBolt(), 2).shuffleGrouping("SpoutBolt");         
        tb.setBolt("CountBolt", new CountBolt(), 4).fieldsGrouping("SplitBolt", new Fields("word"));         
        //创建配置         
        Config conf = new Config();         
        //设置 worker 数量         
        conf.setNumWorkers(2);         
        //提交任务         
        //集群提交         
        //StormSubmitter.submitTopology("myWordcount", conf, tb.createTopology());         
        //本地提交         
        LocalCluster localCluster = new LocalCluster();         
        localCluster.submitTopology("myWordcount", conf, tb.createTopology()); 
    }  
}

代码100分

SplitBolt 部分

代码100分package com.test.stormwordcount;
import java.util.Map; 
import backtype.storm.task.OutputCollector; 
import backtype.storm.task.TopologyContext; 
import backtype.storm.topology.OutputFieldsDeclarer; 
import backtype.storm.topology.base.BaseRichBolt; 
import backtype.storm.tuple.Fields; 
import backtype.storm.tuple.Tuple; 
import backtype.storm.tuple.Values; 

public class SplitBolt extends BaseRichBolt{      
    OutputCollector collector; 

    /**      * 初始化      */     
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {         
        this.collector = collector;     
        } 

    /**      * 执行方法      */     
    public void execute(Tuple input) {         
        String line = input.getString(0);         
        String[] split = line.split(" ");         
        for (String word : split) {             
            collector.emit(new Values(word));         
            }     
        } 

    /**      * 输出      */     
    public void declareOutputFields(OutputFieldsDeclarer declarer) {         
        declarer.declare(new Fields("word"));     
        } 
}

CountBolt 部分

package com.test.stormwordcount;
import java.util.HashMap; 
import java.util.Map; 
import backtype.storm.task.OutputCollector; 
import backtype.storm.task.TopologyContext; 
import backtype.storm.topology.OutputFieldsDeclarer; 
import backtype.storm.topology.base.BaseRichBolt; 
import backtype.storm.tuple.Tuple; 

public class CountBolt extends BaseRichBolt{ 

    OutputCollector collector;
    Map<String, Integer> map = new HashMap<String, Integer>(); 

    /**      * 初始化      */     
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {         
        this.collector = collector;     
        } 


    /**      * 执行方法      */     
public void execute(Tuple input) {         
    String word = input.getString(0);         
    if(map.containsKey(word)){             
    Integer c = map.get(word);             
        map.put(word, c+1);         
        }else{             
        map.put(word, 1);         
        }         
    //测试输出         
    System.out.println("结果:"+map);     
    } 

    /**      * 输出      */     
public void declareOutputFields(OutputFieldsDeclarer declarer) {     
    
} 
}

SpoutBolt 部分

代码100分package com.test.stormwordcount;
import java.util.Map; 
import backtype.storm.spout.SpoutOutputCollector; 
import backtype.storm.task.TopologyContext; 
import backtype.storm.topology.OutputFieldsDeclarer; 
import backtype.storm.topology.base.BaseRichSpout; 
import backtype.storm.tuple.Fields; 
import backtype.storm.tuple.Values; 

public class SpoutBolt extends BaseRichSpout{ 

    SpoutOutputCollector collector;
    /**      * 初始化方法      */     
    public void open(Map map, TopologyContext context, SpoutOutputCollector collector) {         
        this.collector = collector;     
        } 

    /**      * 重复调用方法      */     
    public void nextTuple() {         
        collector.emit(new Values("hello world this is a test"));     
        } 

    /**      * 输出      */     
    public void declareOutputFields(OutputFieldsDeclarer declarer) {         
        declarer.declare(new Fields("test"));     
        } 
}

POM.XML 文件内容

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>

<groupId>com.test</groupId>
<artifactId>stormwordcount</artifactId>
<version>0.9.6</version>
<packaging>jar</packaging>

<name>stormwordcount</name>
<url>http://maven.apache.org</url>

<properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>3.8.1</version>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.apache.storm</groupId>
        <artifactId>storm-core</artifactId>
        <version>0.9.6</version>
    </dependency>
</dependencies>
<build>
    <plugins>
        <plugin>
            <artifactId>maven-assembly-plugin</artifactId>
            <configuration>
                <descriptorRefs>
                    <descriptorRef>jar-with-dependencies</descriptorRef>
                </descriptorRefs>
                <archive>
                    <manifest>
                        <mainClass>com.test.stormwordcount.MainClass</mainClass>
                    </manifest>
                </archive>
            </configuration>
            <executions>
                <execution>
                    <id>make-assembly</id>
                    <phase>package</phase>
                    <goals>
                        <goal>single</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-compiler-plugin</artifactId>
            <configuration>
                <source>1.7</source>
                <target>1.7</target>
            </configuration>
        </plugin>
    </plugins>
</build>

遇到的问题

基于Storm的WordCount需要eclipse安装了maven插件，之前的大数据实践安装的eclipse版本为Eclipse IDE for Eclipse Committers4.5.2，这个版本不自带maven插件，后续安装失败了几次（网上很多的教程都已经失效），这里分享一下我成功安装的方法：
使用链接下载，Help->Install New SoftWare
基于Storm的WordCount「建议收藏」
点击Add，name输入随意，在location输入下载eclipse的maven插件，下载地址可以这样获取
点击连接：http://www.eclipse.org/m2e/index.html 进入网站后点击download，拉到最下面可以看到很多eclipse maven插件的版本和发布时间，选在适合eclipse的版本复制链接即可。建议取消选中Contack all update sites during install to find required software(耗时太久)。

但是安装成功后还是无法配置（这里原因不太清楚，没找到解决办法），就直接上官网换成自己maven插件的JavaEE IDE了…

后续的maven的配置这些都比较顺利，第一次创建maven-archetype-quickstat项目报错，试了网上很多办法都还没成功，然后打开 Windows->Preferencs->Maven->Installation发现之前配置了的maven的安装路径没了…重新配置了下就可以创建项目了。

最后运行成功的结果：
基于Storm的WordCount「建议收藏」

版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 举报，一经查实，本站将立刻删除。
转载请注明出处: https://daima100.com/10423.html

赞 (0)
小小码小小码
0 0
上一篇 2022-12-27 22:00
下一篇 2022-12-28

相关推荐

  • mongodb的坑_踩坑 代码基础

    mongodb的坑_踩坑

    mongodb的坑_踩坑mongo中的日期，在显示上，会比我们正常的时间少8h。 如果向mongo中插入数据，会少8h 如果从mongo中查出数据，那么在idea中会是正常的；而如果是在datagrip/navicat中查，

    小小码 小小码
    2023-06-13
    134
  • sql sever 统计表记录数「建议收藏」 代码基础

    sql sever 统计表记录数「建议收藏」

    sql sever 统计表记录数「建议收藏」select a.name as 表名,max(b.rows) as 记录条数 from FIPIDC.dbo.sysobjects a ,FIPIDC.dbo.sysindexes b where…

    小小码 小小码
    2023-03-14
    133
  • day10-习题[通俗易懂] 代码基础

    day10-习题[通俗易懂]

    day10-习题[通俗易懂]习题 1.Homework01 (1) D 没有在别名上加引号（ps：别名的as可以省略） (2) B 判断null或非空不能用不等于号 (3) C 2.Homework02 写出查看de

    小小码 小小码
    2023-06-10
    116
  • Python A Min：快速高效的数据分析工具 代码基础

    Python A Min：快速高效的数据分析工具

    Python A Min：快速高效的数据分析工具Python一直以来都是数据科学家、数据工程师和分析师的首选编程语言之一。它易于学习，支持大量的第三方包和库，而且最重要的是，Python能够以更快捷、更高效、更灵活的方式进行数据处理和分析。

    admin admin
    2024-03-01
    77
  • Python创建中心列表 代码基础

    Python创建中心列表

    Python创建中心列表Python是一种简单、易学、功能强大的编程语言，它在数据处理、科学计算、Web开发等领域都有广泛的应用。在Python中，可以通过创建列表来存储一组有序的数据，这使得Python具有处理数据的优秀能力。

    admin admin
    2024-04-21
    54
  • mysql创建用户和权限管理目录_软件权限管理 代码基础

    mysql创建用户和权限管理目录_软件权限管理

    mysql创建用户和权限管理目录_软件权限管理转自：https://www.cnblogs.com/zhaohaiyu/p/11459613.html 一、如何创建用户和密码 1.进入到mysql数据库下 mysql> use mysql

    小小码 小小码
    2023-03-17
    128
  • 使用Python re库的基础教程 代码基础

    使用Python re库的基础教程

    使用Python re库的基础教程Python是一门强大的编程语言，能够应用于多种场景，包括数据分析、Web开发、人工智能等，其中正则表达式(re)是Python中非常常用的工具。本文将详细阐述如何使用Python re库，帮助读者更好地掌握正则表达式的应用技巧。

    admin admin
    2024-05-28
    41
  • 用Python求解圆周率的平方根 代码基础

    用Python求解圆周率的平方根

    用Python求解圆周率的平方根求解圆周率的方法有很多种，其中一种常用的方法是蒙特卡罗方法。假设有一个圆和一个正方形，正方形边长为圆的直径，当在正方形中随机生成大量点时，经过圆内点的数量占总点数的比例将趋近于圆的面积与正方形面积的比例，而圆的面积与正方形面积的比例是圆周率的平方的1/4。于是，通过这一比例就可以近似求出圆周率的值。

    admin admin
    2024-03-02
    63

发表回复

您的电子邮箱地址不会被公开。 必填项已用*标注