安科网

kafka-storm 详细

wangying

wangying

2016-12-19

关注 关注

  

 
kafka-storm 详细
 

kafka storm 安装：

15.安装kafka
cd /usr/local/
wget http://mirror.bit.edu.cn/apache/kafka/0.10.0.0/kafka_2.10-0.10.0.0.tgz
tar xf kafka_2.10-0.10.0.0.tgz
ln -s /usr/local/kafka_2.10-0.10.0.0 /usr/local/kafka
chown -R hdfs:hadoop /usr/local/kafka_2.10-0.10.0.0 /usr/local/kafka
chown -R root:root /usr/local/kafka_2.10-0.10.0.0 /usr/local/kafka

/usr/local/zookeeper/bin/zkCli.sh
create /kafka ''

vim /usr/local/kafka/config/server.properties
broker.id=0
zookeeper.connect=dev10.aoiplus.openpf:2181,dev06.aoiplus.openpf:2181,dev05.aoiplus.openpf:2181/kafka

scp -r /usr/local/kafka_2.10-0.10.0.0.tgz [email protected]:/usr/local/
scp -r /usr/local/kafka_2.10-0.10.0.0.tgz [email protected]:/usr/local/

scp -r /usr/local/kafka/config/server.properties [email protected]:/usr/local/kafka/config/server.properties
scp -r /usr/local/kafka/config/server.properties [email protected]:/usr/local/kafka/config/server.properties

master slave 启动
/usr/local/kafka/bin/kafka-server-start.sh /usr/local/kafka/config/server.properties &
创建topic
/usr/local/kafka/bin/kafka-topics.sh --create --zookeeper dev10.aoiplus.openpf:2181,dev06.aoiplus.openpf:2181,dev05.aoiplus.openpf:2181/kafka --replication-factor 3 --partitions 5 --topic baoy-topic
/usr/local/kafka/bin/kafka-topics.sh --describe --zookeeper dev10.aoiplus.openpf:2181,dev06.aoiplus.openpf:2181,dev05.aoiplus.openpf:2181/kafka --topic baoy-topic


/usr/local/kafka/bin/kafka-console-producer.sh --broker-list dev10.aoiplus.openpf:9092,dev05.aoiplus.openpf:9092,dev06.aoiplus.openpf:9092 --topic baoy-topic

/usr/local/kafka/bin/kafka-console-consumer.sh --zookeeper dev10.aoiplus.openpf:2181,dev05.aoiplus.openpf:2181,dev06.aoiplus.openpf:2181/kafka --from-beginning --topic baoy-topic

  

16. storm 安装
cd /usr/local/
wget http://mirrors.cnnic.cn/apache/storm/apache-storm-0.10.0/apache-storm-0.10.0.tar.gz
tar xf apache-storm-0.10.0.tar.gz
ln -s /usr/local/apache-storm-0.10.0 /usr/local/storm
chown -R storm:storm /usr/local/apache-storm-0.10.0 /usr/local/storm
chown -R root:root /usr/local/apache-storm-0.10.0 /usr/local/storm

mkdir -p /tmp/storm/data/

cd storm
vim conf/storm.yaml
	storm.zookeeper.servers:
	     - "dev10.aoiplus.openpf"
	     - "dev05.aoiplus.openpf"
	     - "dev06.aoiplus.openpf"
	storm.zookeeper.port: 2181

	nimbus.host: "dev10.aoiplus.openpf"

	supervisor.slots.ports:
	    - 6700
	    - 6701
	    - 6702
	    - 6703

	storm.local.dir: "/tmp/storm/data"

scp -r /usr/local/storm/conf/storm.yaml [email protected]:/usr/local/storm/conf/
scp -r /usr/local/storm/conf/storm.yaml [email protected]:/usr/local/storm/conf/

启动
master
/usr/local/storm/bin/storm nimbus >/dev/null 2>&1 & 
/usr/local/storm/bin/storm ui >/dev/null 2>&1 & 
slaves
/usr/local/storm/bin/storm supervisor >/dev/null 2>&1 &
查看
http://dev10.aoiplus.openpf/index.html



cp /usr/local/kafka/libs/kafka_2.10-0.10.0.0.jar /usr/local/storm/lib/
cp /usr/local/kafka/libs/scala-library-2.10.6.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.1.2.4.jar /usr/local/storm/lib/
cp /usr/local/kafka/libs/zkclient-0.8.jar /usr/local/storm/lib/
cp /usr/local/kafka/libs/log4j-1.2.17.jar /usr/local/storm/lib/
cp /usr/local/kafka/libs/slf4j-api-1.7.21.jar /usr/local/storm/lib/
cp /usr/local/kafka/libs/jopt-simple-3.2.jar /usr/local/storm/lib/




/usr/local/storm/bin/storm jar /home/baoy/soft/storm/KafkaStormJavaDemo_main_start.jar com.curiousby.baoyou.cn.storm.TerminalInfosAnalysisTopology  "terminalInfosAnalysisTopology" 

mkdir -p /home/baoy/soft/storm/logs
chmod -R 777 /home/baoy/soft/storm/logs

cd /usr/local/storm/log4j2/
vim cluster.xml
<property name="logpath">/home/baoy/soft/storm</property>   

关闭 storm
/usr/local/storm/bin/storm kill terminalInfosAnalysisTopology


kafka-storm 详细
 
kafka-storm 详细
 
kafka-storm 详细
 

pom.xml

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <groupId>com.curiousby.baoy.cn</groupId>
  <artifactId>KafkaStormJavaDemo</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <packaging>jar</packaging>
  <name>SpringKafkaStormDemo</name>
  <url>http://maven.apache.org</url>



	<!-- properties constant -->
	<properties>
		<spring.version>4.2.5.RELEASE</spring.version>
		  <java.version>1.7</java.version>
	</properties>

	<repositories>
		<repository>
			<id>clojars.org</id>
			<url>http://clojars.org/repo</url>
		</repository>
	</repositories>
 

		
		
	<dependencies> 
		<dependency>
			<groupId>org.json</groupId>
			<artifactId>json</artifactId>
			<version>20160810</version>
		</dependency>
		
		<!-- <dependency>
			<groupId>org.slf4j</groupId>
			<artifactId>slf4j-api</artifactId>
			<version>1.6.4</version>
			<type>jar</type>
		</dependency>
		<dependency>
			<groupId>org.slf4j</groupId>
			<artifactId>slf4j-log4j12</artifactId>
			<version>1.6.4</version>
			<type>jar</type>
		</dependency> -->
	 <dependency>  
         <groupId>org.apache.storm</groupId>  
         <artifactId>storm-core</artifactId>  
        <version>0.9.6</version>  
         <scope>provided</scope>  
     </dependency>  
     <dependency>  
        <groupId>org.apache.storm</groupId>  
         <artifactId>storm-kafka</artifactId>  
         <version>0.9.6</version>  
     </dependency>  
     <dependency>  
         <groupId>org.apache.kafka</groupId>  
         <artifactId>kafka_2.10</artifactId>  
         <version>0.9.0.1</version>  
             <exclusions>  
                 <exclusion>  
                     <groupId>org.apache.zookeeper</groupId>  
                     <artifactId>zookeeper</artifactId>  
                </exclusion>  
                 <exclusion>  
                     <groupId>log4j</groupId>  
                     <artifactId>log4j</artifactId>  
                 </exclusion>  
                <exclusion>  
                     <groupId>org.slf4j</groupId>  
                    <artifactId>slf4j-log4j12</artifactId>  
                </exclusion>  
             </exclusions>  
     </dependency>  
	 
 


	</dependencies>
	<build>
		<finalName>SpringKafkaStormDemo</finalName>
		 
		<plugins> 
			<plugin>
				<groupId>org.apache.maven.plugins</groupId>
				<artifactId>maven-compiler-plugin</artifactId>
				<version>3.3</version>
				<dependencies>
					<dependency>
						<groupId>org.codehaus.plexus</groupId>
						<artifactId>plexus-compiler-javac</artifactId>
						<version>2.5</version>
					</dependency>
				</dependencies>
				<configuration>
					<source>1.7</source>
					<target>1.7</target>
					<encoding>UTF-8</encoding>
					<compilerArguments>
						<verbose />
						<bootclasspath>${java.home}/lib/rt.jar:${java.home}/lib/jce.jar</bootclasspath>
					</compilerArguments>
				</configuration>
			</plugin>
		 
			<plugin>
			    <groupId>org.apache.maven.plugins</groupId>
			    <artifactId>maven-shade-plugin</artifactId>
			    <version>1.4</version>
			    <configuration>
			        <appendAssemblyId>false</appendAssemblyId> 
				   <finalName>${project.artifactId}_TerminalInfosAnalysisTopology_main_start</finalName>
			        <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 implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
			                        <mainClass>com.curiousby.baoyou.cn.storm.TerminalInfosAnalysisTopology</mainClass>
			                    </transformer>
			                </transformers>
			            </configuration>
			        </execution>
			    </executions>
			</plugin>
			<plugin>  
              <artifactId>maven-assembly-plugin</artifactId>  
              <configuration>  
                   <appendAssemblyId>false</appendAssemblyId> 
				   <finalName>${project.artifactId}_main_start</finalName>
                  <descriptorRefs>    
                      <descriptorRef>jar-with-dependencies</descriptorRef>  
                  </descriptorRefs>  
                  <archive>  
                     <manifest>  
                       <mainClass>com.curiousby.baoyou.cn.storm.TerminalInfosAnalysisTopology</mainClass>  
                     </manifest>  
                   </archive>  
               </configuration>  
            </plugin> 
			
		</plugins>
	</build>
</project>
package com.curiousby.baoyou.cn.storm;

import java.util.UUID;

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.topology.TopologyBuilder;
import storm.kafka.BrokerHosts;
import storm.kafka.KafkaSpout;
import storm.kafka.SpoutConfig;
import storm.kafka.StringScheme;
import storm.kafka.ZkHosts;

/**
 * @see com.curiousby.baoyou.cn.storm.TerminalInfosAnalysisTopology
 * @Type TerminalInfosAnalysisTopology.java
 * @Desc 
 * @author cmcc-B100036
 * @date 2016年12月15日 下午4:54:50
 * @version 
 */
public class TerminalInfosAnalysisTopology {

    private static String topicName = "baoy-topic";
    private static String zkRoot = "" ;

    public static void main(String[] args) {
        BrokerHosts hosts = new ZkHosts(
                "172.23.27.120:2181,172.23.27.115:2181,172.23.27.116:2181/kafka");
        SpoutConfig spoutConfig = new SpoutConfig(hosts, topicName, zkRoot, UUID.randomUUID().toString());
        spoutConfig.forceFromStart= false;
        spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
        
        KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig);

        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("kafkaSpout", kafkaSpout);
        builder.setBolt("terminalInfosAnalysisRedisBolt", new TerminalInfosAnalysisRedisBolt(),2).shuffleGrouping("kafkaSpout");  
        builder.setBolt("terminalInfosAnalysisElasticsearchBolt", new TerminalInfosAnalysisElasticsearchBolt(), 2).shuffleGrouping("kafkaSpout");  
        
         
        Config conf = new Config();
        conf.setDebug(true);

        if (args != null && args.length > 0) {
            conf.setNumWorkers(2);
            try {
                StormSubmitter.submitTopologyWithProgressBar(args[0], conf,  builder.createTopology());
            } catch (AlreadyAliveException e) {
                e.printStackTrace();
            } catch (InvalidTopologyException e) {
                e.printStackTrace();
            }
        } else { 
            conf.setMaxSpoutPending(2);
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("terminalInfosAnalysisTopology", conf, builder.createTopology());

        }

    }
}
public class TerminalInfosAnalysisRedisBolt extends BaseRichBolt {

    private Logger logger =LoggerFactory.getLogger(TerminalInfosAnalysisRedisBolt.class);
    private OutputCollector collector;
    
    @Override
    public void execute(Tuple tuple) { 
        JSONObject formate = TerminalInfos.formate(tuple.getString(0)); 
        TerminalInfos entity = new TerminalInfos();
        entity.formate(formate);  
        if (entity != null) {
            System.out.println(entity);
            logger.info("===========================================================");
            logger.info("========================="+entity+"=========================");
            logger.info("===========================================================");
        } 
    }

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

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {

    }

}

使用 kafka 客户端 定时发送 


kafka-storm 详细
 

kafka-storm 及时处理 


kafka-storm 详细
 

这里面 我使用的是本地模式

捐助开发者

在兴趣的驱动下,写一个免费的东西，有欣喜，也还有汗水，希望你喜欢我的作品，同时也能支持一下。 当然，有钱捧个钱场（右上角的爱心标志，支持支付宝和PayPal捐助），没钱捧个人场，谢谢各位。


kafka-storm 详细kafka-storm 详细
 
 谢谢您的赞助，我会做的更好！

kafka

wangying

wangying

0 关注 0 粉丝 0 动态

关注 关注

相关推荐

.NET Core下使用Kafka的方法步骤

# 修改每个topic的默认分区参数num.partitions，默认是1，具体合适的取值需要根据服务器配置进程确定，UCloud.ukafka = 3. Segment：partition物理上由多个segment组成，下面2.2和2.3有详细说明。pa

Kafka 2020-09-18

架构师的选择，Pulsar还是Kafka？

最近，我一直在研究Pulsar及其与Kafka的比较。快速搜索将显示两个最著名的开源消息传递系统之间存在当前的"战争"。作为Kafka的用户，我确实对Kafka的某些问题感到困惑，并且我对Pulsar感到非常失望。在本文中，我将重点介绍

yanghuashuiyue 9评论 2020-11-14

Kafka之消费与心跳

导读kafka是一个分布式，分区的，多副本的，多订阅者的消息发布订阅系统，可以用于搜索日志，监控日志，访问日志等。今天小编来领大家一起来学习一下Kafka消费与心跳机制。Kafka提供了非常简单的消费API，使用者只需初始化Kafka的Broker Se

liuxingen 12评论 2020-11-13

刨根问底,Kafka消息中间件到底会不会丢消息

本文转载自微信公众号「爱笑的架构师」，作者雷架。大型互联网公司一般都会要求消息传递最大限度的不丢失，比如用户服务给代金券服务发送一个消息，如果消息丢失会造成用户未收到应得的代金券，最终用户会投诉。为避免上面类似情况的发生，除了做好补偿措施，更应该在系设计的

wangying 2020-11-13

记一次生产 Kafka 挂掉的那几分钟

本文转载自微信公众号「Java极客技术」，作者鸭血粉丝。Hello，大家好，我是阿粉，作为一个后端工程师不经历几次生产事故怎么能成长!阿粉工作几年来，大大小小，重要不重要的事故也经历了不少，有损失几十万的，有对业务毫无影响但是不应该发生的，每一次事故都是一

王谦 15评论 2020-11-03

面试 | 再有人问Kafka，别说你不会！

Kafka 是一个优秀的分布式消息中间件，许多系统中都会使用到 Kafka 来做消息通信。对分布式消息系统的了解和使用几乎成为一个后台开发人员必备的技能。今天就从常见的 Kafka 面试题入手和大家聊聊 Kafka 的那些事儿。生产者负责创建消息，然后将其

huangwei00 10评论 2020-10-14

RabbitMQ七战Kafka，差异立现！

作为一个有丰富经验的微服务系统架构师，经常有人问我，“应该选择RabbitMQ还是Kafka？”。基于某些原因， 许多开发者会把这两种技术当做等价的来看待。的确，在一些案例场景下选择RabbitMQ还是Kafka没什么差别，但是这两种技术在底层实现方面是有

shenzhenzsw 2020-10-09

Kafka突然宕机了？稳住，莫慌！

问题要从一次 Kafka 的宕机开始说起。笔者所在的是一家金融科技公司，但公司内部并没有采用在金融支付领域更为流行的 RabbitMQ，而是采用了设计之初就为日志处理而生的 Kafka，所以我一直很好奇 Kafka 的高可用实现和保障。从 Kafka 部

guicaizhou 2020-09-30

监控kafka队列长度

#!df=pd.DataFrame(columnList,columns=["topics","LAG"]). dfResult=df.groupby("topics",as_index=Fals

jiaomrswang 2020-09-23

Kafka都没整明白，还敢去面试？

放眼当下数据为王的时代，深入了解 Apache Kafka 及其常见的部署应用，快速实现数据架构已是大势所趋，刻不容缓。以下分别 Kafka 架构，四大核心 API，典型应用场景，Kafka 代理与消息主题，集群的创建，流 APIs及其处理模式等不同方面展

jyj0 2020-09-21

SAP Data Intelligence里的Kafka Producer和Consumer

新建一个graph，使用典型的生产者-消费者模型：将Data Generator生成的数据交给kafka Producer operator；而Kafka Consumer从kafka producer里读取出data Generator生成的数据，通过T

guicaizhou 2020-09-15

kafka技术题

将向 Kafka topic 发布消息的程序成为 producers.将预订 topics 并消费消息的程序成为 consumer.Kafka 以集群的方式运行，可以由一个或多个服务组成，每个服务叫做一个 broker.producers 通过网络将消息发

hannuotayouxi 2020-08-20

kafka常见面试题

我们举个例子说明下运维中面对的复杂性，我们都知道 kafka 有个 ISR集合，我先说明下这个概念：

yangyutong00 2020-08-09

Windows 安装kafka

直接去 D:\Tool\zookeeper-3.4.14\bin， 运行 zkServer.cmd. kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1

amwayy 2020-08-03

【kafka】消费组消费主题

当消费组中的消费者数量大于分区数量时，会存在浪费。因为有消费者分配不到partition进行消费。在消费者不饱和的情况下，一个消费者是可以去消费多个分区的数据的 。生产者生产投递9条消息，可以9条消息被投递到3不同分区，然后三个分区对应三个不同的消费者进行

yangyutong00 2020-08-01

logstash_output_kafka:Mysql同步Kafka深入详解

实际业务场景中，会遇到基础数据存在Mysql中，实时写入数据量比较大的情景。迁移至kafka是一种比较好的业务选型方案。如果需要同步历史全量数据+实时更新数据，建议使用logstash。实际上，核心logstash的同步原理的掌握，有助于大家理解类似的各种

weikaixxxxxx 2020-08-01

【kafka】生产者投递消息特征（异步发送）

kafka的生产者producer由持有未提交给kafka服务的记录的缓冲区构成一个缓冲区池，于此同时会有一个后台I/O线程来负责将这些记录record转化为请求request，然后将其提交给集群。如果发送完数据之后不对producer进行资源释放clos

PoppyEvan 2020-08-01

【kafka】消费者对应的分配partition分区策略

解决问题：1.实现了负载均衡。现有两个主题topic1、topic2，各自都有三个partition。TopicAndPartition对象，多个主题整体排序[topic1_partition0,topic1_partition1,topic1_parti

guicaizhou 2020-08-01

Kafka基础（八）：高效读写数据

Kafka 的 producer 生产数据，要写入到 log 文件中，写的过程是一直追加到文件末端，为顺序写。官网有数据表明，同样的磁盘，顺序写能到 600M/s，而随机写只有 100K/s。与磁盘的机械机构有关，顺序写之所以快，是因为其省去了大量磁头寻址

PoppyEvan 2020-07-29

Apache Kafka 架构和相关概念

缓冲上下游瞬时突发流量，使其更平滑.特别是对于那种发送能力很强的上游系统，如果没有消息引擎的保护，“脆弱”的下游系统可能会直接被压垮导致全链路服务“雪崩”。使发送方与接收方松耦合,仅以协议的方式进行通讯,简化了开发.也叫消息队列,每个消息只能被一个下游的消

sweetgirl0 2020-07-27
wangying

wangying

W3CSchool教程
HTML 教程
CSS 教程
Bootstrap 教程
Javascript 教程
jQuery 教程
后端教程
C 教程
Java 教程
PHP 教程
Python 教程
Go 教程
移动开发
Android 教程
Swift 教程
Kotlin 教程
jQuery Mobile 教程
ionic 教程
关于我们
新闻动态
联系方式
招聘英才
安科实验室
帮助与反馈

安科网(Ancii)，中国第一极客网

Copyright © 2013 - 2019 Ancii.com

京ICP备18063983号 京公网安备11010802014868号