程序员经验分享-hwhale

大数据：Storm集成Kafka

2023-10-27

一、整合说明

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 发送过来的数据：

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)

大数据

kafka

storm

大数据：Storm集成Kafka 的相关文章

随机推荐

  • 自建私人图床方案：使用Cpolar+树洞外链轻松部署超轻量级图床，实现高效图片存储

    文章目录 1 前言 2 树洞外链网站搭建 2 1 树洞外链下载和安装 2 2 树洞外链网页测试 2 3 cpolar的安装和注册 3 本地网页发布 3 1 Cpolar临时数据隧道 3 2 Cpolar稳定隧道 云端设置 3 3 Cpola

  • 用python 修改websocketpp头文件格式的时候发现python写csv文件时候有多余的双引号

    def transformer row str level int previous row if level 1 row row replace lt elif level 2 row row replace lt elif level

  • linux中脚本监测nginx状态并重启

    首先要确定nginx的安装位置 如果自定义安装则需要自己去找到nginx运行程序所在的位置 下面的编写和执行都是nginx通过源代码编译安装 运行程序所在的位置为 usr local nginx sbin nginx 确定nginx的使用端

  • mmsegmentation 之修改输入通道

    open mmlab有许多非常实用的框架 其中目标检测的话mmdetection确实很实用 但语义分割的话当属mmsegmentation 这篇博客介绍mmsegmentation如何将输入图像通道数修改为单通道 1 默认你自己已经用mms

  • Stable Diffusion核心算法DDPM解析

    DDPM Denoising Diffusion Probabilistic Model 去噪扩散概率模型 本文参考 一个视频看懂扩散模型DDPM原理推导 AI绘画底层模型 哔哩哔哩 bilibili 1 大概原理 从右往左为正向加噪过程

  • 完美数的算法设计（C语言）

    完全数 Perfect number 又称完美数或完备数 是一些特殊的自然数 它所有的真因子 即除了自身以外的约数 的和 即因子函数 恰好等于它本身 如果一个数恰好等于它的因子之和 则称该数为 完全数 寻找完美的数 题目描述 所谓完美的数是

  • 机器学习——卷积神经网络基础

    卷积神经网络 Convolutional Neural Network CNN 卷积神经网络是人工神经网络的一种 是一种前馈神经网络 最早提出时的灵感来源于人类的神经元 通俗来讲 其主要的操作就是 接受输入层的输入信息 通过卷积层 激活层

  • Pinia刷新会丢失数据 应该怎么解决

    Pinia pinia官网 pinia plugin persistedstate pinia plugin persistedstate官网 Pinia 是一个 Vue js 状态管理库 它允许你创建一个全局的 store 和多个模块化的

  • 不同页面参数值传递localStorage

    不同页面参数值传递localStorage 介绍 localStorage 和 sessionStorage 属性允许在浏览器中存储 key value 对的数据 localStorage 用于长久保存整个网站的数据 保存的数据没有过期时间

  • C语言-qsort函数详解

    一 qsort函数是什么 qsort 函数 快速排序的函数 引用stdlib h头文件 参数说明 void qsort void base 要排序的目标数组 size t num 待排序的元素个数 size t width 一个元素的大小

  • Java最小栈

    Java中的栈是一种后进先出 Last In First Out LIFO 的数据结构 它具有如下几个基本操作 将一个元素压入栈中 称为 入栈 push 将栈顶元素弹出 称为 出栈 pop 查看栈顶元素 称为 查看栈顶元素 peek 判断栈

  • Jenkins详细教程

    最近接触了jenkins这个东西 所以花点时间了解了下 它可以在代码上传仓库 如github gitee gitlab 后 在jenkins 一个网站界面 中通过获取代码仓库中最新代码 进行自动化部署 而省去手动打包 上传服务器 部署这一系

  • 多线程实现TCP的服务器，客户端随时收发信息

    1 作业需求 完成TCP的服务器 客户端 上交 服务器 客户端需要做到随时收发 多进程多线程实现 2 实现过程 1 服务器代码 include

  • 手机影像ISP流程：AWB（1）

    目录 AWB简介 AWB算法 灰度世界算法 灰度世界法的代码 灰度世界法的优点 灰度世界法的缺点 镜面反射法 镜面反射法的代码 镜面反射法的优点 镜面反射法的缺点 总结 AWB简介 AWB是Automatic white balance的英

  • 微信小程序中wx:for语法及其使用

    微信小程序官方文档中wx for 是用来做列表渲染 在组件上使用wx for绑定一个数组 数组的下标变量名默认为index 数组当前项的变量名默认为item 在Wxml中可以使用for循环获取数组数据 记住for循环后面跟的是数组 当然数组

  • 语义分割的常用指标详解

    1 混淆矩阵 假设有6个类别 L为10个真实标签的取值 P为对应的预测的标签值 先计算对应的n 类别数 这里假设为6 xL P bin的值一定是分类数的平方 混淆矩阵先将真实标签和预测标签抻成一维向量 做一个对应关系 nxL P 再将这个对

  • Linux必杀（十六）：打包命令tar

    题记 tar 的参数非常多 挑重点的讲几个 tar j z cv f 新建的文件名 filename lt 打包与压缩 tar j z tv f 新建的文件名 lt 查看文件名 tar j z xv f 新建的文件名 c目录 lt 解压缩

  • JPA @MappedSuperclass注解的使用说明

    原文地址 http hi baidu com danghj item 507d7bdcee3ac052d63aaedf 基于代码复用和模型分离的思想 在项目开发中使用JPA的 MappedSuperclass注解将实体类的多个属性分别封装到

  • win7安装linux子系统,win7系统安装Version 1607中的Linux子系统的方法介绍

    win7系统使用久了 好多网友反馈说win7系统安装Version 1607中的Linux子系统的问题 非常不方便 有什么办法可以永久解决win7系统安装Version 1607中的Linux子系统的问题 面对win7系统安装Version

  • 大数据：Storm集成Kafka

    一 整合说明 Storm 官方对 Kafka 的整合分为两个版本 官方说明文档分别如下 Storm Kafka Integration 主要是针对 0 8 x 版本的 Kafka 提供整合支持 Storm Kafka Integration

热门标签