基於Storm的WordCount

来源:https://www.cnblogs.com/p1ng/archive/2019/12/17/12057174.html
-Advertisement-
Play Games

Storm WordCount 工作過程 Storm 版本： 1、Spout 從外部數據源中讀取數據，隨機發送一個元組對象出去； 2、SplitBolt 接收 Spout 中輸出的元組對象，將元組中的數據切分成單詞，並將切分後的單詞發射出去； 3、WordCountBolt 接收 SplitBolt ...


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()); 
    }  
}

SplitBolt 部分

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 部分

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

點擊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的安裝路徑沒了...重新配置了下就可以創建項目了。

最後運行成功的結果：


您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • inux 資源監控分析-pidstat
    pidstat主要用於監控全部或指定進程占用系統資源的情況，如CPU，記憶體、設備IO、任務切換、線程等。pidstat首次運行時顯示自系統啟動開始的各項統計信息，之後運行pidstat將顯示自上次運行該命令以後的統計信息。用戶可以通過指定統計的次數和時間來獲得所需的統計信息。 ...
  • 網路管理基礎
    網路管理基礎 Network interface names Traditionally, network interfaces in Linux are enumerated as eth0, eth1, eth2, and so on. However, the mechanism which ...
  • 存儲管理
    存儲基礎知識 主要知識點: 基本分區、邏輯捲LVM、EXT3/4/XFS文件系統、RAID從工作原理區分機械固態從磁碟尺寸3.5 2.5 1.8從插拔方式熱插拔非熱插拔從硬碟主要介面IDE —— SATA I/II/IIISCSI —— SAS PCIEFCkernel對不同介面硬碟命名方式OS I ...
  • 進程管理
    關於進程 process What is a process? 什麼是進程Process life cycle 進程的生命周期Process states 進程狀態什麼是進程？進程是已啟動的可執行程式的運行實例，進程有以下組成部分：• 已分配記憶體的地址空間；• 安全屬性，包括所有權憑據和特權；• 程 ...
  • 查看yum已安裝的包
    在linux下如何使用yum查看安裝了哪些軟體包 列出所有已安裝的軟體包 yum list installed yum針對軟體包操作常用命令： 1.使用 yum 查找軟體包 命令：yum search 2.列出所有可安裝的軟體包 命令：yum list 3.列出所有可更新的軟體包 命令：yum li ...
  • 安裝包RPM包或源碼包
    RPM工具 # mount /dev/cdrom /mnt 掛載光碟 # rpm 軟體包管理器 -i 安裝（需要安裝包完整名稱） -v 可視化 -h 顯示安裝進度 -U 表示升級 -q 查詢包是否安裝 -qa 查詢當前系統已安裝的所有包 -qi 查詢已安裝的包的相關信息 -ql 查詢包的安裝文件 - ...
  • 海思HI3518EV200+AR0130開發板DIY——前篇
    一、電源部分 HI3518EV200+AR0130方案電源部分分別有5V/3V3/1V8/1V1。輸入電源USB埠供電5V輸入，5V再通過DCDC降壓轉成3V3，1V8和1V1。1V8和1V1需晚於3V3上電。 電源部分 ...
  • Nginx：基本概念
    守住一方平安，儘力而為，問心無愧就好。 Nginx同Apache一樣都是一種WEB伺服器，Nginx是一款自由的、開源的、高性能的HTTP伺服器和反向代理伺服器；同時也是一個IMAP、POP3、SMTP代理伺服器；Nginx可以作為一個HTTP伺服器進行網站的發佈處理，另外Nginx可以作為反向代理 ...
一周排行
    -Advertisement-
    Play Games
  • 移動開發（一）：使用.NET MAUI開發第一個安卓APP
    移動開發（一）：使用.NET MAUI開發第一個安卓APP 對於工作多年的C#程式員來說，近來想嘗試開發一款安卓APP，考慮了很久最終選擇使用.NET MAUI這個微軟官方的框架來嘗試體驗開發安卓APP，畢竟是使用Visual Studio開發工具，使用起來也比較的順手，結合微軟官方的教程進行了安卓 ...
  • wpf ToggleButton選中效果和一個登錄界面
    前言 QuestPDF 是一個開源 .NET 庫，用於生成 PDF 文檔。使用了C# Fluent API方式可簡化開發、減少錯誤並提高工作效率。利用它可以輕鬆生成 PDF 報告、發票、導出文件等。 項目介紹 QuestPDF 是一個革命性的開源 .NET 庫，它徹底改變了我們生成 PDF 文檔的方 ...
  • 彈幕樹洞項目功能新增篇
    項目地址 項目後端地址： https://github.com/ZyPLJ/ZYTteeHole 項目前端頁面地址： ZyPLJ/TreeHoleVue (github.com) https://github.com/ZyPLJ/TreeHoleVue 目前項目測試訪問地址： http://tree ...
  • 第27篇 sqlserver2022詳細安裝步驟
    話不多說，直接開乾 一.下載 1.官方鏈接下載： https://www.microsoft.com/zh-cn/sql-server/sql-server-downloads 2.在下載目錄中找到下麵這個小的安裝包 SQL2022-SSEI-Dev.exe，運行開始下載SQL server； 二. ...
  • .NET 開源高性能 MQTT 類庫
    前言 隨著物聯網（IoT）技術的迅猛發展，MQTT（消息隊列遙測傳輸）協議憑藉其輕量級和高效性，已成為眾多物聯網應用的首選通信標準。 MQTTnet 作為一個高性能的 .NET 開源庫，為 .NET 平臺上的 MQTT 客戶端與伺服器開發提供了強大的支持。 本文將全面介紹 MQTTnet 的核心功能 ...
  • Serilog文檔翻譯系列（六） - 可用的接收器、增強器、格式化輸出
    Serilog支持多種接收器用於日誌存儲，增強器用於添加屬性，LogContext管理動態屬性，支持多種輸出格式包括純文本、JSON及ExpressionTemplate。還提供了自定義格式化選項，適用於不同需求。 ...
  • 警惕 Visual Studio 屬性求值副作用導致邏輯不符合預期
    目錄簡介獲取 HTML 文檔解析 HTML 文檔測試參考文章 簡介 動態內容網站使用 JavaScript 腳本動態檢索和渲染數據，爬取信息時需要模擬瀏覽器行為，否則獲取到的源碼基本是空的。 本文使用的爬取步驟如下： 使用 Selenium 獲取渲染後的 HTML 文檔 使用 HtmlAgility ...
  • [使用目前最新版]HybridCLR6.9.0+YooAsset2.2.4實現純C# Unity熱更新方案 (一)
    1.前言 什麼是熱更新 游戲或者軟體更新時，無需重新下載客戶端進行安裝，而是在應用程式啟動的情況下，在內部進行資源或者代碼更新 Unity目前常用熱更新解決方案 HybridCLR，Xlua，ILRuntime等 Unity目前常用資源管理解決方案 AssetBundles，Addressable， ...
  • 在 ASP.NET Core Web API 中使用操作篩選器統一處理通用操作
    本文章主要是在C# ASP.NET Core Web API框架實現向手機發送驗證碼簡訊功能。這裡我選擇是一個互億無線簡訊驗證碼平臺，其實像阿裡雲，騰訊雲上面也可以。 首先我們先去 互億無線 https://www.ihuyi.com/api/sms.html 去註冊一個賬號 註冊完成賬號後，它會送 ...
  • 第28篇 如何.net中實現高效可靠數據同步api
    通過以下方式可以高效，並保證數據同步的可靠性 1.API設計 使用RESTful設計，確保API端點明確，並使用適當的HTTP方法（如POST用於創建，PUT用於更新）。 設計清晰的請求和響應模型，以確保客戶端能夠理解預期格式。 2.數據驗證 在伺服器端進行嚴格的數據驗證，確保接收到的數據符合預期格 ...
所有分類