Storm 系列（九）—— Storm 集成 Kafka
程序员文章站
2022-05-03 15:02:46
一、整合说明 Storm 官方对 Kafka 的整合分为两个版本，官方说明文档分别如下： + "Storm Kafka Integration" : 主要是针对 0.8.x 版本的 Kafka 提供整合支持； + "Storm Kafka Integration (0.10.x+)" : 包含 Ka ......
一、整合说明
storm 官方对 kafka 的整合分为两个版本，官方说明文档分别如下：
- storm kafka integration : 主要是针对 0.8.x 版本的 kafka 提供整合支持；
- storm kafka integration (0.10.x+) : 包含 kafka 新版本的 consumer api，主要对 kafka 0.10.x + 提供整合支持。
这里我服务端安装的 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
参考资料
更多大数据系列文章可以参见 github 开源项目： 大数据入门指南