Storm 系列（九）—— Storm 整合 Kafka
阿新 • • 發佈：2019-09-05
一、整合說明
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 Storm2.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.properties2. 建立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:90923. 啟動消費者
啟動一個消費者用於觀察寫入情況，啟動命令如下：
# bin/kafka-console-consumer.sh --bootstrap-server hadoop001:9092 --topic storm-topic --from-beginning2.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
參考資料
- Storm Kafka Integration (0.10.x+)
更多大資料系列文章可以參見 GitHub 開源專案： 大資料入門指南