Storm 系列（九）—— Storm 集成 Kafka

来源:https://www.cnblogs.com/heibaiying/archive/2019/09/05/11462899.html
-Advertisement-
Play Games

一、整合說明 Storm 官方對 Kafka 的整合分為兩個版本，官方說明文檔分別如下： + "Storm Kafka Integration" : 主要是針對 0.8.x 版本的 Kafka 提供整合支持； + "Storm Kafka Integration (0.10.x+)" : 包含 Ka ...


一、整合說明

Storm 官方對 Kafka 的整合分為兩個版本，官方說明文檔分別如下：

這裡我服務端安裝的 Kafka 版本為 2.2.0(Released Mar 22, 2019) ，按照官方 0.10.x+ 的整合文檔進行整合，不適用於 0.8.x 版本的 Kafka。

二、寫入數據到Kafka

2.1 項目結構

2.2 項目主要依賴

<properties>
    <storm.version>1.2.2</storm.version>
    <kafka.version>2.2.0</kafka.version>
</properties>

<dependencies>
    <dependency>
        <groupId>org.apache.storm</groupId>
        <artifactId>storm-core</artifactId>
        <version>${storm.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.storm</groupId>
        <artifactId>storm-kafka-client</artifactId>
        <version>${storm.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>${kafka.version}</version>
    </dependency>
</dependencies>

2.3 DataSourceSpout

/**
 * 產生詞頻樣本的數據源
 */
public class DataSourceSpout extends BaseRichSpout {

    private List<String> list = Arrays.asList("Spark", "Hadoop", "HBase", "Storm", "Flink", "Hive");

    private SpoutOutputCollector spoutOutputCollector;

    @Override
    public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
        this.spoutOutputCollector = spoutOutputCollector;
    }

    @Override
    public void nextTuple() {
        // 模擬產生數據
        String lineData = productData();
        spoutOutputCollector.emit(new Values(lineData));
        Utils.sleep(1000);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("line"));
    }


    /**
     * 模擬數據
     */
    private String productData() {
        Collections.shuffle(list);
        Random random = new Random();
        int endIndex = random.nextInt(list.size()) % (list.size()) + 1;
        return StringUtils.join(list.toArray(), "\t", 0, endIndex);
    }

}

產生的模擬數據格式如下：

Spark   HBase
Hive    Flink   Storm   Hadoop  HBase   Spark
Flink
HBase   Storm
HBase   Hadoop  Hive    Flink
HBase   Flink   Hive    Storm
Hive    Flink   Hadoop
HBase   Hive
Hadoop  Spark   HBase   Storm

2.4 WritingToKafkaApp

/**
 * 寫入數據到 Kafka 中
 */
public class WritingToKafkaApp {

    private static final String BOOTSTRAP_SERVERS = "hadoop001:9092";
    private static final String TOPIC_NAME = "storm-topic";

    public static void main(String[] args) {


        TopologyBuilder builder = new TopologyBuilder();

        // 定義 Kafka 生產者屬性
        Properties props = new Properties();
        /*
         * 指定 broker 的地址清單，清單里不需要包含所有的 broker 地址，生產者會從給定的 broker 里查找其他 broker 的信息。
         * 不過建議至少要提供兩個 broker 的信息作為容錯。
         */
        props.put("bootstrap.servers", BOOTSTRAP_SERVERS);
        /*
         * acks 參數指定了必須要有多少個分區副本收到消息，生產者才會認為消息寫入是成功的。
         * acks=0 : 生產者在成功寫入消息之前不會等待任何來自伺服器的響應。
         * acks=1 : 只要集群的首領節點收到消息，生產者就會收到一個來自伺服器成功響應。
         * acks=all : 只有當所有參與複製的節點全部收到消息時，生產者才會收到一個來自伺服器的成功響應。
         */
        props.put("acks", "1");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        KafkaBolt bolt = new KafkaBolt<String, String>()
                .withProducerProperties(props)
                .withTopicSelector(new DefaultTopicSelector(TOPIC_NAME))
                .withTupleToKafkaMapper(new FieldNameBasedTupleToKafkaMapper<>());

        builder.setSpout("sourceSpout", new DataSourceSpout(), 1);
        builder.setBolt("kafkaBolt", bolt, 1).shuffleGrouping("sourceSpout");


        if (args.length > 0 && args[0].equals("cluster")) {
            try {
                StormSubmitter.submitTopology("ClusterWritingToKafkaApp", new Config(), builder.createTopology());
            } catch (AlreadyAliveException | InvalidTopologyException | AuthorizationException e) {
                e.printStackTrace();
            }
        } else {
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("LocalWritingToKafkaApp",
                    new Config(), builder.createTopology());
        }
    }
}

2.5 測試準備工作

進行測試前需要啟動 Kakfa：

1. 啟動Kakfa

Kafka 的運行依賴於 zookeeper，需要預先啟動，可以啟動 Kafka 內置的 zookeeper,也可以啟動自己安裝的：

# zookeeper啟動命令
bin/zkServer.sh start

# 內置zookeeper啟動命令
bin/zookeeper-server-start.sh config/zookeeper.properties

啟動單節點 kafka 用於測試：

# bin/kafka-server-start.sh config/server.properties

2. 創建topic

# 創建用於測試主題
bin/kafka-topics.sh --create --bootstrap-server hadoop001:9092 --replication-factor 1 --partitions 1 --topic storm-topic

# 查看所有主題
 bin/kafka-topics.sh --list --bootstrap-server hadoop001:9092

3. 啟動消費者

啟動一個消費者用於觀察寫入情況，啟動命令如下：

# bin/kafka-console-consumer.sh --bootstrap-server hadoop001:9092 --topic storm-topic --from-beginning

2.6 測試

可以用直接使用本地模式運行，也可以打包後提交到伺服器集群運行。本倉庫提供的源碼預設採用 maven-shade-plugin 進行打包，打包命令如下：

# mvn clean package -D maven.test.skip=true

啟動後，消費者監聽情況如下：

三、從Kafka中讀取數據

3.1 項目結構

3.2 ReadingFromKafkaApp

/**
 * 從 Kafka 中讀取數據
 */
public class ReadingFromKafkaApp {

    private static final String BOOTSTRAP_SERVERS = "hadoop001:9092";
    private static final String TOPIC_NAME = "storm-topic";

    public static void main(String[] args) {

        final TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("kafka_spout", new KafkaSpout<>(getKafkaSpoutConfig(BOOTSTRAP_SERVERS, TOPIC_NAME)), 1);
        builder.setBolt("bolt", new LogConsoleBolt()).shuffleGrouping("kafka_spout");

        // 如果外部傳參 cluster 則代表線上環境啟動,否則代表本地啟動
        if (args.length > 0 && args[0].equals("cluster")) {
            try {
                StormSubmitter.submitTopology("ClusterReadingFromKafkaApp", new Config(), builder.createTopology());
            } catch (AlreadyAliveException | InvalidTopologyException | AuthorizationException e) {
                e.printStackTrace();
            }
        } else {
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("LocalReadingFromKafkaApp",
                    new Config(), builder.createTopology());
        }
    }

    private static KafkaSpoutConfig<String, String> getKafkaSpoutConfig(String bootstrapServers, String topic) {
        return KafkaSpoutConfig.builder(bootstrapServers, topic)
                // 除了分組 ID,以下配置都是可選的。分組 ID 必須指定,否則會拋出 InvalidGroupIdException 異常
                .setProp(ConsumerConfig.GROUP_ID_CONFIG, "kafkaSpoutTestGroup")
                // 定義重試策略
                .setRetry(getRetryService())
                // 定時提交偏移量的時間間隔,預設是 15s
                .setOffsetCommitPeriodMs(10_000)
                .build();
    }

    // 定義重試策略
    private static KafkaSpoutRetryService getRetryService() {
        return new KafkaSpoutRetryExponentialBackoff(TimeInterval.microSeconds(500),
                TimeInterval.milliSeconds(2), Integer.MAX_VALUE, TimeInterval.seconds(10));
    }
}

3.3 LogConsoleBolt

/**
 * 列印從 Kafka 中獲取的數據
 */
public class LogConsoleBolt extends BaseRichBolt {


    private OutputCollector collector;

    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        this.collector=collector;
    }

    public void execute(Tuple input) {
        try {
            String value = input.getStringByField("value");
            System.out.println("received from kafka : "+ value);
            // 必須 ack,否則會重覆消費 kafka 中的消息
            collector.ack(input);
        }catch (Exception e){
            e.printStackTrace();
            collector.fail(input);
        }

    }

    public void declareOutputFields(OutputFieldsDeclarer declarer) {

    }
}

這裡從 value 欄位中獲取 kafka 輸出的值數據。

在開發中，我們可以通過繼承 RecordTranslator 介面定義了 Kafka 中 Record 與輸出流之間的映射關係，可以在構建 KafkaSpoutConfig 的時候通過構造器或者 setRecordTranslator() 方法傳入，並最後傳遞給具體的 KafkaSpout。

預設情況下使用內置的 DefaultRecordTranslator，其源碼如下，FIELDS 中 定義了 tuple 中所有可用的欄位：主題，分區，偏移量，消息鍵，值。

public class DefaultRecordTranslator<K, V> implements RecordTranslator<K, V> {
    private static final long serialVersionUID = -5782462870112305750L;
    public static final Fields FIELDS = new Fields("topic", "partition", "offset", "key", "value");
    @Override
    public List<Object> apply(ConsumerRecord<K, V> record) {
        return new Values(record.topic(),
                record.partition(),
                record.offset(),
                record.key(),
                record.value());
    }

    @Override
    public Fields getFieldsFor(String stream) {
        return FIELDS;
    }

    @Override
    public List<String> streams() {
        return DEFAULT_STREAM;
    }
}

3.4 啟動測試

這裡啟動一個生產者用於發送測試數據，啟動命令如下：

# bin/kafka-console-producer.sh --broker-list hadoop001:9092 --topic storm-topic

本地運行的項目接收到從 Kafka 發送過來的數據：


用例源碼下載地址：storm-kafka-integration

參考資料

  1. Storm Kafka Integration (0.10.x+)

更多大數據系列文章可以參見 GitHub 開源項目： 大數據入門指南


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

-Advertisement-
Play Games
更多相關文章
  • MySQL——python交互
    與python交互之前我們需要安裝一個MySQL的驅動模塊Connector，這個驅動模塊直接在cmd命令行輸入 安裝是否成功可以接著輸入 進入自己的編輯器（pycharm）首先連接資料庫方式： 資料庫連接 1、當前模塊直接連接 2、導入配置文件進行連接 游標 游標的作用是執行sql語句，或者保存s ...
  • Drop Table對MySQL的性能影響分析
    【作者】 王棟：攜程技術保障中心資料庫專家，對資料庫疑難問題的排查和資料庫自動化智能化運維工具的開發有強烈的興趣。 【問題描述】 最近碰到有台MySQL實例出現了MySQL服務短暫hang死，表現為瞬間的併發線程上升，連接數暴增。 排查Error Log文件中有page_cleaner超時的信息，引 ...
  • 大數據應用常用打包方式
    一、簡介 在提交大數據作業到集群上運行時，通常需要先將項目打成 JAR 包。這裡以 Maven 為例，常用打包方式如下： 不加任何插件，直接使用 mvn package 打包； 使用 maven assembly plugin 插件； 使用 maven shade plugin 插件； 使用 mav ...
  • ResourceManager學習之Application,狀態機管理
    Application管理 YARN中,Application是指應用程式,他可能啟動多個運行實例,每個運行實例由一個ApplicationMaster與一組該ApplicationMaster啟動的任務組成,他擁有名稱、隊列名、優先順序等屬性,是一個比較寬泛的概念,可以是一個MapReduce作業、 ...
  • SQL資料庫基礎語法
    SQL語句的概述 SQL語言的分類 數據定義語言（Data　Definition　Language）主要用於修改、創建和刪除資料庫對象，其中包括CREATE ALTER DROP語句。 數據查詢語言（Data Query Language）主要用於查詢資料庫中的數據，其主要是SELECT語句，SEL ...
  • Redis和MongoDB區別
    MongoDB 更類似 MySQL，支持欄位索引、游標操作，其優勢在於查詢功能比較強大，擅長查詢 JSON 數據，能存儲海量數據，但是不支持事務。 Redis 是一個開源（BSD許可）的，記憶體中的數據結構存儲系統，支持多種類型的數據結構，可用作資料庫，高速緩存和消息隊列代理。 1、記憶體管理機制 Re ...
  • navicat遠程連接mysql錯誤
    確保埠與服務正常。 如果都正常，那麼服務有可能沒有添加遠程訪問。 本人鏈接報錯10060，那麼問題出在阿裡雲的主機預設不開放3306埠，那麼在安全組策略中加入對應的埠 登錄後輸入 use mysql，選擇使用mysql資料庫，因為修改遠程連接的基本信息保存在mysql資料庫中，所以使用mysq ...
  • 從零搭建企業大數據分析和機器學習平臺-技術棧介紹（三）
    數據傳輸和採集 Sqoop數據傳輸工具 實際項目開發中，往往很多業務數據是存放在關係型資料庫中，如 MySQL資料庫。我們需要將這些數據集中到數據倉庫中進行管理，便於使用計算模型進行統計、挖掘這類操作。 Sqoop是Apache軟體基金會的⼀一款頂級開源數據傳輸工具，用於在 Hadoop與關係型數據 ...
一周排行
    -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.數據驗證 在伺服器端進行嚴格的數據驗證，確保接收到的數據符合預期格 ...
所有分類