Kafka+Storm+HDFS整合实践-MySQL 튜토리얼-php.cn

집

데이터 베이스

MySQL 튜토리얼

Kafka+Storm+HDFS整合实践

WBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWB

Jun 07, 2016 pm 04:39 PM

기반으로 관행 통합

在基于Hadoop平台的很多应用场景中，我们需要对数据进行离线和实时分析，离线分析可以很容易地借助于Hive来实现统计分析，但是对于实时的需求Hive就不合适了。实时应用场景可以使用Storm，它是一个实时处理系统，它为实时处理类应用提供了一个计算模型，可以

在基于Hadoop平台的很多应用场景中，我们需要对数据进行离线和实时分析，离线分析可以很容易地借助于Hive来实现统计分析，但是对于实时的需求Hive就不合适了。实时应用场景可以使用Storm，它是一个实时处理系统，它为实时处理类应用提供了一个计算模型，可以很容易地进行编程处理。为了统一离线和实时计算，一般情况下，我们都希望将离线和实时计算的数据源的集合统一起来作为输入，然后将数据的流向分别经由实时系统和离线分析系统，分别进行分析处理，这时我们可以考虑将数据源（如使用Flume收集日志）直接连接一个消息中间件，如Kafka，可以整合Flume+Kafka，Flume作为消息的Producer，生产的消息数据（日志数据、业务请求数据等等）发布到Kafka中，然后通过订阅的方式，使用Storm的Topology作为消息的Consumer，在Storm集群中分别进行如下两个需求场景的处理：

直接使用Storm的Topology对数据进行实时分析处理
整合Storm+HDFS，将消息处理后写入HDFS进行离线分析处理

实时处理，只要开发满足业务需要的Topology即可，不做过多说明。这里，我们主要从安装配置Kafka、Storm，以及整合Kafka+Storm、整合Storm+HDFS、整合Kafka+Storm+HDFS这几点来配置实践，满足上面提出的一些需求。配置实践使用的软件包如下所示：

zookeeper-3.4.5.tar.gz
kafka_2.9.2-0.8.1.1.tgz
apache-storm-0.9.2-incubating.tar.gz
hadoop-2.2.0.tar.gz

程序配置运行所基于的操作系统为CentOS 5.11。

Kafka安装配置

我们使用3台机器搭建Kafka集群：

192.168.4.142   h1
192.168.4.143   h2
192.168.4.144   h3

로그인 후 복사

在安装Kafka集群之前，这里没有使用Kafka自带的Zookeeper，而是独立安装了一个Zookeeper集群，也是使用这3台机器，保证Zookeeper集群正常运行。
首先，在h1上准备Kafka安装文件，执行如下命令：

cd /usr/local/
wget http://mirror.bit.edu.cn/apache/kafka/0.8.1.1/kafka_2.9.2-0.8.1.1.tgz
tar xvzf kafka_2.9.2-0.8.1.1.tgz
ln -s /usr/local/kafka_2.9.2-0.8.1.1 /usr/local/kafka
chown -R kafka:kafka /usr/local/kafka_2.9.2-0.8.1.1 /usr/local/kafka

로그인 후 복사

修改配置文件/usr/local/kafka/config/server.properties，修改如下内容：

broker.id=0
zookeeper.connect=h1:2181,h2:2181,h3:2181/kafka

로그인 후 복사

这里需要说明的是，默认Kafka会使用ZooKeeper默认的/路径，这样有关Kafka的ZooKeeper配置就会散落在根路径下面，如果你有其他的应用也在使用ZooKeeper集群，查看ZooKeeper中数据可能会不直观，所以强烈建议指定一个chroot路径，直接在zookeeper.connect配置项中指定：

zookeeper.connect=h1:2181,h2:2181,h3:2181/kafka

로그인 후 복사

而且，需要手动在ZooKeeper中创建路径/kafka，使用如下命令连接到任意一台ZooKeeper服务器：

cd /usr/local/zookeeper
bin/zkCli.sh

로그인 후 복사

在ZooKeeper执行如下命令创建chroot路径：

create /kafka ''

로그인 후 복사

这样，每次连接Kafka集群的时候（使用--zookeeper选项），也必须使用带chroot路径的连接字符串，后面会看到。
然后，将配置好的安装文件同步到其他的h2、h3节点上：

scp -r /usr/local/kafka_2.9.2-0.8.1.1/ h2:/usr/local/
scp -r /usr/local/kafka_2.9.2-0.8.1.1/ h3:/usr/local/

로그인 후 복사

最后，在h2、h3节点上配置，执行如下命令：

cd /usr/local/
ln -s /usr/local/kafka_2.9.2-0.8.1.1 /usr/local/kafka
chown -R kafka:kafka /usr/local/kafka_2.9.2-0.8.1.1 /usr/local/kafka

로그인 후 복사

并修改配置文件/usr/local/kafka/config/server.properties内容如下所示：

broker.id=1  # 在h1修改
broker.id=2  # 在h2修改

로그인 후 복사

因为Kafka集群需要保证各个Broker的id在整个集群中必须唯一，需要调整这个配置项的值（如果在单机上，可以通过建立多个Broker进程来模拟分布式的Kafka集群，也需要Broker的id唯一，还需要修改一些配置目录的信息）。
在集群中的h1、h2、h3这三个节点上分别启动Kafka，分别执行如下命令：

bin/kafka-server-start.sh /usr/local/kafka/config/server.properties &

로그인 후 복사

可以通过查看日志，或者检查进程状态，保证Kafka集群启动成功。
我们创建一个名称为my-replicated-topic5的Topic，5个分区，并且复制因子为3，执行如下命令：

bin/kafka-topics.sh --create --zookeeper h1:2181,h2:2181,h3:2181/kafka --replication-factor 3 --partitions 5 --topic my-replicated-topic5

로그인 후 복사

查看创建的Topic，执行如下命令：

bin/kafka-topics.sh --describe --zookeeper h1:2181,h2:2181,h3:2181/kafka --topic my-replicated-topic5

로그인 후 복사

结果信息如下所示：

Topic:my-replicated-topic5     PartitionCount:5     ReplicationFactor:3     Configs:
     Topic: my-replicated-topic5     Partition: 0     Leader: 0     Replicas: 0,2,1     Isr: 0,2,1
     Topic: my-replicated-topic5     Partition: 1     Leader: 0     Replicas: 1,0,2     Isr: 0,2,1
     Topic: my-replicated-topic5     Partition: 2     Leader: 2     Replicas: 2,1,0     Isr: 2,0,1
     Topic: my-replicated-topic5     Partition: 3     Leader: 0     Replicas: 0,1,2     Isr: 0,2,1
     Topic: my-replicated-topic5     Partition: 4     Leader: 2     Replicas: 1,2,0     Isr: 2,0,1

로그인 후 복사

上面Leader、Replicas、Isr的含义如下：

Partition： 分区
Leader   ： 负责读写指定分区的节点
Replicas ： 复制该分区log的节点列表
Isr      ： "in-sync" replicas，当前活跃的副本列表（是一个子集），并且可能成为Leader

로그인 후 복사

我们可以通过Kafka自带的bin/kafka-console-producer.sh和bin/kafka-console-consumer.sh脚本，来验证演示如果发布消息、消费消息。
在一个终端，启动Producer，并向我们上面创建的名称为my-replicated-topic5的Topic中生产消息，执行如下脚本：

bin/kafka-console-producer.sh --broker-list h1:9092,h2:9092,h3:9092 --topic my-replicated-topic5

로그인 후 복사

在另一个终端，启动Consumer，并订阅我们上面创建的名称为my-replicated-topic5的Topic中生产的消息，执行如下脚本：

bin/kafka-console-consumer.sh --zookeeper h1:2181,h2:2181,h3:2181/kafka --from-beginning --topic my-replicated-topic5

로그인 후 복사

可以在Producer终端上输入字符串消息行，然后回车，就可以在Consumer终端上看到消费者消费的消息内容。
也可以参考Kafka的Producer和Consumer的Java API，通过API编码的方式来实现消息生产和消费的处理逻辑。

Storm安装配置

Storm集群也依赖Zookeeper集群，要保证Zookeeper集群正常运行。Storm的安装配置比较简单，我们仍然使用下面3台机器搭建：

192.168.4.142   h1
192.168.4.143   h2
192.168.4.144   h3

로그인 후 복사

首先，在h1节点上，执行如下命令安装：

cd /usr/local/
wget http://mirror.bit.edu.cn/apache/incubator/storm/apache-storm-0.9.2-incubating/apache-storm-0.9.2-incubating.tar.gz
tar xvzf apache-storm-0.9.2-incubating.tar.gz
ln -s /usr/local/apache-storm-0.9.2-incubating /usr/local/storm
chown -R storm:storm /usr/local/apache-storm-0.9.2-incubating /usr/local/storm

로그인 후 복사

然后，修改配置文件conf/storm.yaml，内容如下所示：

 storm.zookeeper.servers:
     - "h1"
     - "h2"
     - "h3"
storm.zookeeper.port: 2181
#
nimbus.host: "h1"
supervisor.slots.ports:
    - 6700
    - 6701
    - 6702
    - 6703
storm.local.dir: "/tmp/storm"

로그인 후 복사

将配置好的安装文件，分发到其他节点上：

scp -r /usr/local/apache-storm-0.9.2-incubating/ h2:/usr/local/
scp -r /usr/local/apache-storm-0.9.2-incubating/ h3:/usr/local/

로그인 후 복사

最后，在h2、h3节点上配置，执行如下命令：

cd /usr/local/
ln -s /usr/local/apache-storm-0.9.2-incubating /usr/local/storm
chown -R storm:storm /usr/local/apache-storm-0.9.2-incubating /usr/local/storm

로그인 후 복사

Storm集群的主节点为Nimbus，从节点为Supervisor，我们需要在h1上启动Nimbus服务，在从节点h2、h3上启动Supervisor服务：

bin/storm nimbus &
bin/storm supervisor &

로그인 후 복사

为了方便监控，可以启动Storm UI，可以从Web页面上监控Storm Topology的运行状态，例如在h2上启动：

bin/storm ui &

로그인 후 복사

这样可以通过访问http://h2:8080/来查看Topology的运行状况。

整合Kafka+Storm

消息通过各种方式进入到Kafka消息中间件，比如可以通过使用Flume来收集日志数据，然后在Kafka中路由暂存，然后再由实时计算程序Storm做实时分析，这时我们就需要将在Storm的Spout中读取Kafka中的消息，然后交由具体的Spot组件去分析处理。实际上，apache-storm-0.9.2-incubating这个版本的Storm已经自带了一个集成Kafka的外部插件程序storm-kafka，可以直接使用，例如我使用的Maven依赖配置，如下所示：

          <dependency>
               <groupid>org.apache.storm</groupid>
               <artifactid>storm-core</artifactid>
               <version>0.9.2-incubating</version>
               <scope>provided</scope>
          </dependency>
          <dependency>
               <groupid>org.apache.storm</groupid>
               <artifactid>storm-kafka</artifactid>
               <version>0.9.2-incubating</version>
          </dependency>
          <dependency>
               <groupid>org.apache.kafka</groupid>
               <artifactid>kafka_2.9.2</artifactid>
               <version>0.8.1.1</version>
               <exclusions>
                    <exclusion>
                         <groupid>org.apache.zookeeper</groupid>
                         <artifactid>zookeeper</artifactid>
                    </exclusion>
                    <exclusion>
                         <groupid>log4j</groupid>
                         <artifactid>log4j</artifactid>
                    </exclusion>
               </exclusions>
          </dependency>

로그인 후 복사

下面，我们开发了一个简单WordCount示例程序，从Kafka读取订阅的消息行，通过空格拆分出单个单词，然后再做词频统计计算，实现的Topology的代码，如下所示：

package org.shirdrn.storm.examples;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import storm.kafka.BrokerHosts;
import storm.kafka.KafkaSpout;
import storm.kafka.SpoutConfig;
import storm.kafka.StringScheme;
import storm.kafka.ZkHosts;
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.spout.SchemeAsMultiScheme;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
public class MyKafkaTopology {
     public static class KafkaWordSplitter extends BaseRichBolt {
          private static final Log LOG = LogFactory.getLog(KafkaWordSplitter.class);
          private static final long serialVersionUID = 886149197481637894L;
          private OutputCollector collector;
          @Override
          public void prepare(Map stormConf, TopologyContext context,
                    OutputCollector collector) {
               this.collector = collector;              
          }
          @Override
          public void execute(Tuple input) {
               String line = input.getString(0);
               LOG.info("RECV[kafka -> splitter] " + line);
               String[] words = line.split("\\s+");
               for(String word : words) {
                    LOG.info("EMIT[splitter -> counter] " + word);
                    collector.emit(input, new Values(word, 1));
               }
               collector.ack(input);
          }
          @Override
          public void declareOutputFields(OutputFieldsDeclarer declarer) {
               declarer.declare(new Fields("word", "count"));         
          }
     }
     public static class WordCounter extends BaseRichBolt {
          private static final Log LOG = LogFactory.getLog(WordCounter.class);
          private static final long serialVersionUID = 886149197481637894L;
          private OutputCollector collector;
          private Map<string atomicinteger> counterMap;
          @Override
          public void prepare(Map stormConf, TopologyContext context,
                    OutputCollector collector) {
               this.collector = collector;    
               this.counterMap = new HashMap<string atomicinteger>();
          }
          @Override
          public void execute(Tuple input) {
               String word = input.getString(0);
               int count = input.getInteger(1);
               LOG.info("RECV[splitter -> counter] " + word + " : " + count);
               AtomicInteger ai = this.counterMap.get(word);
               if(ai == null) {
                    ai = new AtomicInteger();
                    this.counterMap.put(word, ai);
               }
               ai.addAndGet(count);
               collector.ack(input);
               LOG.info("CHECK statistics map: " + this.counterMap);
          }
          @Override
          public void cleanup() {
               LOG.info("The final result:");
               Iterator<entry atomicinteger>> iter = this.counterMap.entrySet().iterator();
               while(iter.hasNext()) {
                    Entry<string atomicinteger> entry = iter.next();
                    LOG.info(entry.getKey() + "\t:\t" + entry.getValue().get());
               }
          }
          @Override
          public void declareOutputFields(OutputFieldsDeclarer declarer) {
               declarer.declare(new Fields("word", "count"));         
          }
     }
     public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException, InterruptedException {
          String zks = "h1:2181,h2:2181,h3:2181";
          String topic = "my-replicated-topic5";
          String zkRoot = "/storm"; // default zookeeper root configuration for storm
          String id = "word";
          BrokerHosts brokerHosts = new ZkHosts(zks);
          SpoutConfig spoutConf = new SpoutConfig(brokerHosts, topic, zkRoot, id);
          spoutConf.scheme = new SchemeAsMultiScheme(new StringScheme());
          spoutConf.forceFromStart = false;
          spoutConf.zkServers = Arrays.asList(new String[] {"h1", "h2", "h3"});
          spoutConf.zkPort = 2181;
          TopologyBuilder builder = new TopologyBuilder();
          builder.setSpout("kafka-reader", new KafkaSpout(spoutConf), 5); // Kafka我们创建了一个5分区的Topic，这里并行度设置为5
          builder.setBolt("word-splitter", new KafkaWordSplitter(), 2).shuffleGrouping("kafka-reader");
          builder.setBolt("word-counter", new WordCounter()).fieldsGrouping("word-splitter", new Fields("word"));
          Config conf = new Config();
          String name = MyKafkaTopology.class.getSimpleName();
          if (args != null && args.length > 0) {
               // Nimbus host name passed from command line
               conf.put(Config.NIMBUS_HOST, args[0]);
               conf.setNumWorkers(3);
               StormSubmitter.submitTopologyWithProgressBar(name, conf, builder.createTopology());
          } else {
               conf.setMaxTaskParallelism(3);
               LocalCluster cluster = new LocalCluster();
               cluster.submitTopology(name, conf, builder.createTopology());
               Thread.sleep(60000);
               cluster.shutdown();
          }
     }
}
</string></entry></string></string>

로그인 후 복사

上面程序，在本地调试（使用LocalCluster）不需要输入任何参数，提交到实际集群中运行时，需要传递一个参数，该参数为Nimbus的主机名称。
通过Maven构建，生成一个包含依赖的single jar文件（不要把Storm的依赖包添加进去），例如storm-examples-0.0.1-SNAPSHOT.jar，在提交Topology程序到Storm集群之前，因为用到了Kafka，需要拷贝一下依赖jar文件到Storm集群中的lib目录下面：

cp /usr/local/kafka/libs/kafka_2.9.2-0.8.1.1.jar /usr/local/storm/lib/
cp /usr/local/kafka/libs/scala-library-2.9.2.jar /usr/local/storm/lib/
cp /usr/local/kafka/libs/metrics-core-2.2.0.jar /usr/local/storm/lib/
cp /usr/local/kafka/libs/snappy-java-1.0.5.jar /usr/local/storm/lib/
cp /usr/local/kafka/libs/zkclient-0.3.jar /usr/local/storm/lib/
cp /usr/local/kafka/libs/log4j-1.2.15.jar /usr/local/storm/lib/
cp /usr/local/kafka/libs/slf4j-api-1.7.2.jar /usr/local/storm/lib/
cp /usr/local/kafka/libs/jopt-simple-3.2.jar /usr/local/storm/lib/

로그인 후 복사

然后，就可以提交我们开发的Topology程序了：

bin/storm jar /home/storm/storm-examples-0.0.1-SNAPSHOT.jar org.shirdrn.storm.examples.MyKafkaTopology h1

로그인 후 복사

可以通过查看日志文件（logs/目录下）或者Storm UI来监控Topology的运行状况。如果程序没有错误，可以使用前面我们使用的Kafka Producer来生成消息，就能看到我们开发的Storm Topology能够实时接收到并进行处理。
上面Topology实现代码中，有一个很关键的配置对象SpoutConfig，配置属性如下所示：

spoutConf.forceFromStart = false;

로그인 후 복사

该配置是指，如果该Topology因故障停止处理，下次正常运行时是否从Spout对应数据源Kafka中的该订阅Topic的起始位置开始读取，如果forceFromStart=true，则之前处理过的Tuple还要重新处理一遍，否则会从上次处理的位置继续处理，保证Kafka中的Topic数据不被重复处理，是在数据源的位置进行状态记录。

整合Storm+HDFS

Storm实时计算集群从Kafka消息中间件中消费消息，有实时处理需求的可以走实时处理程序，还有需要进行离线分析的需求，如写入到HDFS进行分析。下面实现了一个Topology，代码如下所示：

package org.shirdrn.storm.examples;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;
import java.util.Random;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.storm.hdfs.bolt.HdfsBolt;
import org.apache.storm.hdfs.bolt.format.DefaultFileNameFormat;
import org.apache.storm.hdfs.bolt.format.DelimitedRecordFormat;
import org.apache.storm.hdfs.bolt.format.FileNameFormat;
import org.apache.storm.hdfs.bolt.format.RecordFormat;
import org.apache.storm.hdfs.bolt.rotation.FileRotationPolicy;
import org.apache.storm.hdfs.bolt.rotation.TimedRotationPolicy;
import org.apache.storm.hdfs.bolt.rotation.TimedRotationPolicy.TimeUnit;
import org.apache.storm.hdfs.bolt.sync.CountSyncPolicy;
import org.apache.storm.hdfs.bolt.sync.SyncPolicy;
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.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import backtype.storm.utils.Utils;
public class StormToHDFSTopology {
     public static class EventSpout extends BaseRichSpout {
          private static final Log LOG = LogFactory.getLog(EventSpout.class);
          private static final long serialVersionUID = 886149197481637894L;
          private SpoutOutputCollector collector;
          private Random rand;
          private String[] records;
          @Override
          public void open(Map conf, TopologyContext context,
                    SpoutOutputCollector collector) {
               this.collector = collector;    
               rand = new Random();
               records = new String[] {
                         "10001     ef2da82d4c8b49c44199655dc14f39f6     4.2.1     HUAWEI G610-U00     HUAWEI     2     70:72:3c:73:8b:22     2014-10-13 12:36:35",
                         "10001     ffb52739a29348a67952e47c12da54ef     4.3     GT-I9300     samsung     2     50:CC:F8:E4:22:E2     2014-10-13 12:36:02",
                         "10001     ef2da82d4c8b49c44199655dc14f39f6     4.2.1     HUAWEI G610-U00     HUAWEI     2     70:72:3c:73:8b:22     2014-10-13 12:36:35"
               };
          }
          @Override
          public void nextTuple() {
               Utils.sleep(1000);
               DateFormat df = new SimpleDateFormat("yyyy-MM-dd_HH-mm-ss");
               Date d = new Date(System.currentTimeMillis());
               String minute = df.format(d);
               String record = records[rand.nextInt(records.length)];
               LOG.info("EMIT[spout -> hdfs] " + minute + " : " + record);
               collector.emit(new Values(minute, record));
          }
          @Override
          public void declareOutputFields(OutputFieldsDeclarer declarer) {
               declarer.declare(new Fields("minute", "record"));         
          }
     }
     public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException, InterruptedException {
          // use "|" instead of "," for field delimiter
          RecordFormat format = new DelimitedRecordFormat()
                  .withFieldDelimiter(" : ");
          // sync the filesystem after every 1k tuples
          SyncPolicy syncPolicy = new CountSyncPolicy(1000);
          // rotate files 
          FileRotationPolicy rotationPolicy = new TimedRotationPolicy(1.0f, TimeUnit.MINUTES);
          FileNameFormat fileNameFormat = new DefaultFileNameFormat()
                  .withPath("/storm/").withPrefix("app_").withExtension(".log");
          HdfsBolt hdfsBolt = new HdfsBolt()
                  .withFsUrl("hdfs://h1:8020")
                  .withFileNameFormat(fileNameFormat)
                  .withRecordFormat(format)
                  .withRotationPolicy(rotationPolicy)
                  .withSyncPolicy(syncPolicy);
          TopologyBuilder builder = new TopologyBuilder();
          builder.setSpout("event-spout", new EventSpout(), 3);
          builder.setBolt("hdfs-bolt", hdfsBolt, 2).fieldsGrouping("event-spout", new Fields("minute"));
          Config conf = new Config();
          String name = StormToHDFSTopology.class.getSimpleName();
          if (args != null && args.length > 0) {
               conf.put(Config.NIMBUS_HOST, args[0]);
               conf.setNumWorkers(3);
               StormSubmitter.submitTopologyWithProgressBar(name, conf, builder.createTopology());
          } else {
               conf.setMaxTaskParallelism(3);
               LocalCluster cluster = new LocalCluster();
               cluster.submitTopology(name, conf, builder.createTopology());
               Thread.sleep(60000);
               cluster.shutdown();
          }
     }
}

로그인 후 복사

上面的处理逻辑，可以对HdfsBolt进行更加详细的配置，如FileNameFormat、SyncPolicy、FileRotationPolicy（可以设置在满足什么条件下，切出一个新的日志，如可以指定多长时间切出一个新的日志文件，可以指定一个日志文件大小达到设置值后，再写一个新日志文件），更多设置可以参考storm-hdfs，。
上面代码在打包的时候，需要注意，使用storm-starter自带的Maven打包配置，可能在将Topology部署运行的时候，会报错，可以使用maven-shade-plugin这个插件，如下配置所示：

               <plugin>
                   <groupid>org.apache.maven.plugins</groupid>
                   <artifactid>maven-shade-plugin</artifactid>
                   <version>1.4</version>
                   <configuration>
                       <createdependencyreducedpom>true</createdependencyreducedpom>
                   </configuration>
                   <executions>
                       <execution>
                           <phase>package</phase>
                           <goals>
                               <goal>shade</goal>
                           </goals>
                           <configuration>
                               <transformers>
                                   <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"></transformer>
                                   <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                       <mainclass></mainclass>
                                   </transformer>
                               </transformers>
                           </configuration>
                       </execution>
                   </executions>
               </plugin>

로그인 후 복사

整合Kafka+Storm+HDFS

上面分别对整合Kafka+Storm和Storm+HDFS做了实践，可以将后者的Spout改成前者的Spout，从Kafka中消费消息，在Storm中可以做简单处理，然后将数据写入HDFS，最后可以在Hadoop平台上对数据进行离线分析处理。下面，写了一个简单的例子，从Kafka消费消息，然后经由Storm处理，写入到HDFS存储，代码如下所示：

package org.shirdrn.storm.examples;
import java.util.Arrays;
import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.storm.hdfs.bolt.HdfsBolt;
import org.apache.storm.hdfs.bolt.format.DefaultFileNameFormat;
import org.apache.storm.hdfs.bolt.format.DelimitedRecordFormat;
import org.apache.storm.hdfs.bolt.format.FileNameFormat;
import org.apache.storm.hdfs.bolt.format.RecordFormat;
import org.apache.storm.hdfs.bolt.rotation.FileRotationPolicy;
import org.apache.storm.hdfs.bolt.rotation.TimedRotationPolicy;
import org.apache.storm.hdfs.bolt.rotation.TimedRotationPolicy.TimeUnit;
import org.apache.storm.hdfs.bolt.sync.CountSyncPolicy;
import org.apache.storm.hdfs.bolt.sync.SyncPolicy;
import storm.kafka.BrokerHosts;
import storm.kafka.KafkaSpout;
import storm.kafka.SpoutConfig;
import storm.kafka.StringScheme;
import storm.kafka.ZkHosts;
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.spout.SchemeAsMultiScheme;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
public class DistributeWordTopology {
     public static class KafkaWordToUpperCase extends BaseRichBolt {
          private static final Log LOG = LogFactory.getLog(KafkaWordToUpperCase.class);
          private static final long serialVersionUID = -5207232012035109026L;
          private OutputCollector collector;
          @Override
          public void prepare(Map stormConf, TopologyContext context,
                    OutputCollector collector) {
               this.collector = collector;              
          }
          @Override
          public void execute(Tuple input) {
               String line = input.getString(0).trim();
               LOG.info("RECV[kafka -> splitter] " + line);
               if(!line.isEmpty()) {
                    String upperLine = line.toUpperCase();
                    LOG.info("EMIT[splitter -> counter] " + upperLine);
                    collector.emit(input, new Values(upperLine, upperLine.length()));
               }
               collector.ack(input);
          }
          @Override
          public void declareOutputFields(OutputFieldsDeclarer declarer) {
               declarer.declare(new Fields("line", "len"));         
          }
     }
     public static class RealtimeBolt extends BaseRichBolt {
          private static final Log LOG = LogFactory.getLog(KafkaWordToUpperCase.class);
          private static final long serialVersionUID = -4115132557403913367L;
          private OutputCollector collector;
          @Override
          public void prepare(Map stormConf, TopologyContext context,
                    OutputCollector collector) {
               this.collector = collector;              
          }
          @Override
          public void execute(Tuple input) {
               String line = input.getString(0).trim();
               LOG.info("REALTIME: " + line);
               collector.ack(input);
          }
          @Override
          public void declareOutputFields(OutputFieldsDeclarer declarer) {
          }
     }
     public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException, InterruptedException {
          // Configure Kafka
          String zks = "h1:2181,h2:2181,h3:2181";
          String topic = "my-replicated-topic5";
          String zkRoot = "/storm"; // default zookeeper root configuration for storm
          String id = "word";
          BrokerHosts brokerHosts = new ZkHosts(zks);
          SpoutConfig spoutConf = new SpoutConfig(brokerHosts, topic, zkRoot, id);
          spoutConf.scheme = new SchemeAsMultiScheme(new StringScheme());
          spoutConf.forceFromStart = false;
          spoutConf.zkServers = Arrays.asList(new String[] {"h1", "h2", "h3"});
          spoutConf.zkPort = 2181;
          // Configure HDFS bolt
          RecordFormat format = new DelimitedRecordFormat()
                  .withFieldDelimiter("\t"); // use "\t" instead of "," for field delimiter
          SyncPolicy syncPolicy = new CountSyncPolicy(1000); // sync the filesystem after every 1k tuples
          FileRotationPolicy rotationPolicy = new TimedRotationPolicy(1.0f, TimeUnit.MINUTES); // rotate files
          FileNameFormat fileNameFormat = new DefaultFileNameFormat()
                  .withPath("/storm/").withPrefix("app_").withExtension(".log"); // set file name format
          HdfsBolt hdfsBolt = new HdfsBolt()
                  .withFsUrl("hdfs://h1:8020")
                  .withFileNameFormat(fileNameFormat)
                  .withRecordFormat(format)
                  .withRotationPolicy(rotationPolicy)
                  .withSyncPolicy(syncPolicy);
          // configure & build topology
          TopologyBuilder builder = new TopologyBuilder();
          builder.setSpout("kafka-reader", new KafkaSpout(spoutConf), 5);
          builder.setBolt("to-upper", new KafkaWordToUpperCase(), 3).shuffleGrouping("kafka-reader");
          builder.setBolt("hdfs-bolt", hdfsBolt, 2).shuffleGrouping("to-upper");
          builder.setBolt("realtime", new RealtimeBolt(), 2).shuffleGrouping("to-upper");
          // submit topology
          Config conf = new Config();
          String name = DistributeWordTopology.class.getSimpleName();
          if (args != null && args.length > 0) {
               String nimbus = args[0];
               conf.put(Config.NIMBUS_HOST, nimbus);
               conf.setNumWorkers(3);
               StormSubmitter.submitTopologyWithProgressBar(name, conf, builder.createTopology());
          } else {
               conf.setMaxTaskParallelism(3);
               LocalCluster cluster = new LocalCluster();
               cluster.submitTopology(name, conf, builder.createTopology());
               Thread.sleep(60000);
               cluster.shutdown();
          }
     }
}

로그인 후 복사

上面代码中，名称为to-upper的Bolt将接收到的字符串行转换成大写以后，会将处理过的数据向后面的hdfs-bolt、realtime这两个Bolt各发一份拷贝，然后由这两个Bolt分别根据实际需要（实时/离线）单独处理。
打包后，在Storm集群上部署并运行这个Topology：

bin/storm jar ~/storm-examples-0.0.1-SNAPSHOT.jar org.shirdrn.storm.examples.DistributeWordTopology h1

로그인 후 복사

可以通过Storm UI查看Topology运行情况，可以查看HDFS上生成的数据。

参考链接

http://kafka.apache.org/
http://kafka.apache.org/documentation.html
https://cwiki.apache.org/confluence/display/KAFKA/Consumer+Group+Example
http://storm.apache.org/
http://storm.apache.org/documentation/Tutorial.html
http://storm.apache.org/documentation/FAQ.html
https://github.com/ptgoetz/storm-hdfs

原文地址：Kafka+Storm+HDFS整合实践, 感谢原作者分享。

본 웹사이트의 성명

본 글의 내용은 네티즌들의 자발적인 기여로 작성되었으며, 저작권은 원저작자에게 있습니다. 본 사이트는 이에 상응하는 법적 책임을 지지 않습니다. 표절이나 침해가 의심되는 콘텐츠를 발견한 경우 admin@php.cn으로 문의하세요.

핫 AI 도구

Undresser.AI Undress

사실적인 누드 사진을 만들기 위한 AI 기반 앱

AI Clothes Remover

사진에서 옷을 제거하는 온라인 AI 도구입니다.

Undress AI Tool

무료로 이미지를 벗다

Clothoff.io

AI 옷 제거제

AI Hentai Generator

AI Hentai를 무료로 생성하십시오.

뜨거운 도구

메모장++7.3.1

사용하기 쉬운 무료 코드 편집기

SublimeText3 중국어 버전

중국어 버전, 사용하기 매우 쉽습니다.

스튜디오 13.0.1 보내기

강력한 PHP 통합 개발 환경

드림위버 CS6

시각적 웹 개발 도구

SublimeText3 Mac 버전

신 수준의 코드 편집 소프트웨어(SublimeText3)

뜨거운 주제

Gmail 이메일의 로그인 입구는 어디에 있나요?

7464

Cakephp 튜토리얼

1376

Steam의 계정 이름 형식은 무엇입니까?

Win11 활성화 키 영구

NYT 연결 힌트와 답변

Related knowledge

Outlook이 내 일정에 이벤트를 자동으로 추가하는 것을 중지하는 방법 Feb 26, 2024 am 09:49 AM

이메일 관리자 애플리케이션인 Microsoft Outlook을 사용하면 이벤트와 약속을 예약할 수 있습니다. 이를 통해 Outlook 응용 프로그램에서 이러한 활동(이벤트라고도 함)을 생성, 관리 및 추적할 수 있는 도구를 제공하여 체계적으로 정리할 수 있습니다. 그러나 때로는 원치 않는 이벤트가 Outlook의 일정에 추가되어 사용자에게 혼란을 주고 일정에 스팸을 보내는 경우가 있습니다. 이 문서에서는 Outlook이 내 일정에 이벤트를 자동으로 추가하지 못하도록 방지하는 데 도움이 되는 다양한 시나리오와 단계를 살펴보겠습니다. Outlook 이벤트 – 간략한 개요 Outlook 이벤트는 다양한 용도로 사용되며 다음과 같은 유용한 기능을 많이 가지고 있습니다. 일정 통합: Outlook에서

Dreamweaver CMS 스테이션 그룹 실습 공유 Mar 18, 2024 am 10:18 AM

Dreamweaver CMS 스테이션 그룹 실습 공유 최근 몇 년간 인터넷의 급속한 발전으로 인해 웹사이트 구축이 점점 더 중요해지고 있습니다. 여러 웹사이트를 구축할 때 사이트 그룹 기술은 매우 효과적인 방법이 되었습니다. 많은 웹 사이트 구축 도구 중에서 DreamWeaver CMS는 유연성과 사용 용이성으로 인해 많은 웹 사이트 애호가들의 첫 번째 선택이 되었습니다. 이 기사에서는 Dreamweaver CMS 스테이션 그룹에 대한 몇 가지 실제 경험과 일부 특정 코드 예제를 공유하여 스테이션 그룹 기술을 탐색하는 독자에게 도움이 되기를 바랍니다. 1. Dreamweaver CMS 스테이션 그룹이란 무엇입니까? 드림위버 CMS

PHP 코딩 방법: Goto 문에 대한 대안 거부 Mar 28, 2024 pm 09:24 PM

PHP 코딩 방법: Goto 문에 대한 대안 사용 거부 최근 몇 년간 프로그래밍 언어의 지속적인 업데이트와 반복으로 인해 프로그래머는 코딩 사양과 모범 사례에 더 많은 관심을 기울이기 시작했습니다. PHP 프로그래밍에서 goto 문은 오랫동안 제어 흐름 문으로 존재해 왔지만, 실제 응용에서는 코드의 가독성과 유지 관리성이 떨어지는 경우가 많습니다. 이 기사에서는 개발자가 goto 문 사용을 거부하고 코드 품질을 향상시키는 데 도움이 되는 몇 가지 대안을 공유합니다. 1. goto 문 사용을 거부하는 이유는 무엇입니까? 먼저 그 이유를 생각해 보자.

Struts 프레임워크의 원칙과 실무에 대한 심층적인 토론 Feb 18, 2024 pm 06:10 PM

Struts 프레임워크의 원리 분석 및 실제 탐색 JavaWeb 개발에서 일반적으로 사용되는 MVC 프레임워크인 Struts 프레임워크는 우수한 디자인 패턴과 확장성을 가지며 엔터프라이즈 수준 애플리케이션 개발에 널리 사용됩니다. 이 기사에서는 Struts 프레임워크의 원리를 분석하고 실제 코드 예제를 통해 이를 탐색하여 독자가 프레임워크를 더 잘 이해하고 적용할 수 있도록 돕습니다. 1. Struts 프레임워크의 원리 분석 1. MVC 아키텍처 Struts 프레임워크는 MVC(Model-View-Con)를 기반으로 합니다.

C++ 반사 메커니즘 실습: 유연한 런타임 유형 정보 구현 Nov 27, 2023 pm 01:11 PM

C++ 리플렉션 메커니즘 실습: 유연한 런타임 유형 정보 구현 소개: C++는 강력한 형식의 언어이며 다른 언어처럼 클래스 유형 정보를 얻기 위한 리플렉션 메커니즘을 직접 제공하지 않습니다. 그러나 몇 가지 트릭과 기술적 수단을 사용하면 C++에서도 유사한 반사 기능을 구현할 수도 있습니다. 이 문서에서는 템플릿 메타프로그래밍과 매크로 정의를 활용하여 유연한 런타임 유형 정보를 얻는 방법을 설명합니다. 1. 반사 메커니즘은 무엇입니까? 리플렉션 메커니즘은 클래스 이름, 멤버 함수, 멤버 변수 및 기타 속성과 같은 클래스의 유형 정보를 런타임에 얻는 것을 의미합니다.

Golang을 사용한 트래픽 관리 모범 사례 Mar 07, 2024 am 08:27 AM

Golang은 웹 서비스 및 애플리케이션을 구축하는 데 널리 사용되는 강력하고 효율적인 프로그래밍 언어입니다. 네트워크 서비스에서 트래픽 관리는 네트워크상의 데이터 전송을 제어 및 최적화하고 서비스의 안정성과 성능을 보장하는 데 도움이 되는 중요한 부분입니다. 이 기사에서는 Golang을 사용한 트래픽 관리 모범 사례를 소개하고 구체적인 코드 예제를 제공합니다. 1. 기본 트래픽 관리를 위해 Golang의 넷 패키지를 사용합니다. Golang의 넷 패키지는 네트워크 데이터를 처리하는 방법을 제공합니다.

PyCharm을 사용한 원격 개발에 대한 실무 가이드 Feb 25, 2024 pm 07:18 PM

원격 개발에 PyCharm을 사용하는 것은 개발자가 로컬 환경의 원격 서버에서 코드를 쉽게 편집, 디버그 및 실행할 수 있도록 하는 효율적인 방법입니다. 이 기사에서는 원격 개발 실습에 PyCharm을 사용하는 방법을 소개하고 이를 특정 코드 예제와 결합하여 독자가 이 기술을 더 잘 이해하고 적용할 수 있도록 돕습니다. PyCharm이란 무엇입니까?PyCharm은 JetBrains에서 개발한 Python 통합 개발 환경(IDE)으로, 이를 지원하는 풍부한 기능과 도구를 제공합니다.

C++ 반사 메커니즘 실습: 유연한 런타임 유형 정보 구현 Nov 27, 2023 pm 01:11 PM

C++ 리플렉션 메커니즘 연습: 유연한 런타임 유형 정보 구현 소개: C++는 강력한 형식의 언어이며 다른 언어처럼 클래스 유형 정보를 얻기 위한 리플렉션 메커니즘을 직접 제공하지 않습니다. 그러나 몇 가지 트릭과 기술적 수단을 사용하면 C++에서도 유사한 반사 기능을 구현할 수도 있습니다. 이 문서에서는 템플릿 메타프로그래밍과 매크로 정의를 활용하여 유연한 런타임 유형 정보를 얻는 방법을 설명합니다. 1. 반사 메커니즘은 무엇입니까? 리플렉션 메커니즘은 클래스 이름, 멤버 함수, 멤버 변수 및 기타 속성과 같은 클래스의 유형 정보를 런타임에 얻는 것을 의미합니다.

See all articles

Kafka+Storm+HDFS整合实践

핫 AI 도구

Undresser.AI Undress

AI Clothes Remover

Undress AI Tool

Clothoff.io

AI Hentai Generator

인기 기사

뜨거운 도구

메모장++7.3.1

SublimeText3 중국어 버전

스튜디오 13.0.1 보내기

드림위버 CS6

SublimeText3 Mac 버전

뜨거운 주제