zoukankan      html  css  js  c++  java

  • Storm应用系列之——集成Kafka

    本文系原创系列，转载请注明。

    原帖地址：http://blog.csdn.net/xeseo

    前言

    在前面Storm系列之——基本概念一文中，提到过Storm的Spout应该是源源不断的取数据，不能间断。那么，很显然，消息队列系统、分布式内存系统或内存数据库是作为其数据源的很好的选择。本文就如何集成Kafka进行介绍。

    Kafka的基本介绍：http://blog.csdn.net/xeseo/article/details/18311955

    准备工作

    KafkaSpout其实网上已经有人写了，在github上开源了，不用我们自己造轮子。只是要注意版本问题：

    0.7版本的Kafka，对应KafkaSpout可以使用Storm-contrib下面的例子

    源码：https://github.com/nathanmarz/storm-contrib/tree/master/storm-kafka

    Maven依赖：https://clojars.org/storm/storm-kafka

    0.8版本的Kafka在API上和底层Offset的处理方式上发生了重大变化，所以老的KafkaSpout不再适用，必须使用新的KafkaAPI

    源码：https://github.com/wurstmeister/storm-kafka-0.8-plus

    Maven依赖：https://clojars.org/net.wurstmeister.storm/storm-kafka-0.8-plus

    这里因为0.8版本的Kafka必然是将来主流，所以我就不介绍0.7 的了，使用方式基本上是类似的。

    PS:

    是人写的，就会有bug，何况是别人分享出来的。所以，遇到bug，还请去github上提交一个issue告诉作者修正。

    2014/7/29 更新：

    wurstmeister/storm-kafka-0.8-plus 现在合并到Apache Storm了，在其external/storm-kakfa目录

    Maven依赖直接更新成：

    [plain] view plaincopyprint?在CODE上查看代码片派生到我的代码片

    1.               <dependency>  

    2.   <groupId>org.apache.storm</groupId>  

    3.   <artifactId>storm-kafka</artifactId>  

    4.   <version>0.9.2-incubating</version>  

    5. </dependency>  

    但是storm似乎没有直接把external的包加载到classpath，所以使用时，还得手动把该jar包从external/storm-kafka/下拷到storm的lib目录。

    当然，也可以在maven中加上<scope>compile</scope>，直接把该jar打到你项目一起。

    使用KafkaSpout

    一个KafkaSpout只能去处理一个topic的内容，所以，它要求初始化时提供如下与topic相关信息：

    • Kafka集群中的Broker地址 （IP+Port）

    有两种方法指定：

    1. 使用静态地址，即直接给定Kafka集群中所有Broker信息

    [java] view plaincopyprint?在CODE上查看代码片派生到我的代码片

    1. GlobalPartitionInformation info = new GlobalPartitionInformation();  

    2. info.addPartition(0, new Broker("10.1.110.24",9092));  

    3. info.addPartition(0, new Broker("10.1.110.21",9092));  

    4. BrokerHosts brokerHosts = new StaticHosts(info);  

    2. 从Zookeeper动态读取

    [java] view plaincopyprint?在CODE上查看代码片派生到我的代码片

    1. BrokerHosts brokerHosts = new ZkHosts("10.1.110.24:2181,10.1.110.22:2181");  

    推荐使用这种方法，因为Kafka的Broker可能会动态的增减

    • topic名字

    • 当前spout的唯一标识Id （以下代称$spout_id）

    • zookeeper上用于存储当前处理到哪个Offset了 （以下代称$zk_root)

    • 当前topic中数据如何解码

    了解Kafka的应该知道，Kafka中当前处理到哪的Offset是由客户端自己管理的。所以，后面两个的目的，其实是在zookeeper上建立一个 $zk_root/$spout_id 的节点，其值是一个map，存放了当前Spout处理的Offset的信息。

    在Topology中加入Spout的代码：

    [java] view plaincopyprint?在CODE上查看代码片派生到我的代码片

    1. String topic = "test";  

    2. String zkRoot = "kafkastorm";  

    3. String spoutId = "myKafka";  

    4.   

    5. SpoutConfig spoutConfig = new SpoutConfig(brokerHosts, topic, zkRoot, spoutId);  

    6. spoutConfig.scheme = new SchemeAsMultiScheme(new TestMessageScheme());  

    7.   

    8. TopologyBuilder builder = new TopologyBuilder();  

    9. builder.setSpout("spout", new KafkaSpout(spoutConfig), spoutNum);  


    其中TestMessageScheme就是告诉KafkaSpout如何去解码数据，生成Storm内部传递数据

    [java] view plaincopyprint?在CODE上查看代码片派生到我的代码片

    1. public class TestMessageScheme implements Scheme {  

    2.   

    3.     private static final Logger LOGGER = LoggerFactory.getLogger(TestMessageScheme.class);  

    4.       

    5.     @Override  

    6.     public List<Object> deserialize(byte[] bytes) {  

    7.         try {  

    8.             String msg = new String(bytes, "UTF-8");  

    9.             return new Values(msg);  

    10.         } catch (InvalidProtocolBufferException e) {  

    11.             LOGGER.error("Cannot parse the provided message!");  

    12.         }  

    13.           

    14.         //TODO: what happend if returns null?  

    15.         return null;  

    16.     }  

    17.   

    18.     @Override  

    19.     public Fields getOutputFields() {  

    20.         return new Fields("msg");  

    21.     }  

    22.   

    23. }  

    这个解码方式是与Producer端生成时塞入数据的编码方式配套的。这里我Producer端塞入的是String的byte，所以这里也还原成String，定义输出为一个名叫"msg"的field。

    后面就可以自己添加Bolt处理tuple中该field的数据了。

    使用TransactionalTridentKafkaSpout

    TransactionalTridentKafkaSpout是为事务性的Trident而用的。用法与KafkaSpout有所不同。

    [java] view plaincopyprint?在CODE上查看代码片派生到我的代码片

    1. TridentKafkaConfig kafkaConfig = new TridentKafkaConfig(brokerHosts, topic, spoutId);  

    2. kafkaConfig.scheme = new SchemeAsMultiScheme(new TestMessageScheme());  

    3.   

    4. TransactionalTridentKafkaSpout kafkaSpout = new TransactionalTridentKafkaSpout(kafkaConfig);  

    5.   

    6. TridentTopology topology = new TridentTopology();  

    7. topology.newStream("test_str", kafkaSpout).shuffle().each(new Fields("msg", new PrintFunction());  

    看到它并没有要求我们提供zkRoot，因为直接代码里面写死了…… -_-T

    地址是 /transactional/<STREAM_NAME>/<Spout_Id>,在上面的例子中，就是  /transactional/test_str/myKafaka

    常见问题

    1. 本地模式无法保存Offset

    KafkaSpout初始化时，会去取spoutConfig.zkServers 和 spoutConfig.zkPort 变量的值，而该值默认是没塞的，所以是空，那么它就会去取当前运行的Storm所配置的zookeeper地址和端口，而本地运行的Storm，是一个临时的zookeeper实例，并不会真正持久化。所以，每次关闭后，数据就没了。

    本地模式，要显示的去配置

    [java] view plaincopyprint?在CODE上查看代码片派生到我的代码片

    1. spoutConfig.zkServers = new ArrayList<String>(){{  

    2.                 add("10.1.110.20");  

    3.                 add("10.1.110.21");  

    4.                 add("10.1.110.24");  

    5.             }};  

    6. spoutConfig.zkPort = 2181;  



    2. 用Maven导入时，运行中SLF4J打印MutipleBinding 错误，导致无log输出。

    原因是在这个KafkaSpout的pom.xml里依赖了kafka_2.9.2,而这货带了一个slf4j-simple的SLF4J绑定，修复这个问题

    [html] view plaincopyprint?在CODE上查看代码片派生到我的代码片

    1. <del><dependency>  

    2.   <groupId>net.wurstmeister.storm</groupId>  

    3.   <artifactId>storm-kafka-0.8-plus</artifactId>  

    4.   <version>0.2.0</version>  

    5.   <exclusion>  

    6.     <groupId>org.slf4j</groupId>  

    7.     <artifactId>slf4j-simple</artifactId>  

    8.   </exclusion>  

    9. </dependency></del>  


    3. 如果在topology第一次启动前，往kafka里面写数据，启动Storm后，这部分数据读不出来

    原因是第一次启动topology时，在zookeeper上并未创建出保存Offset信息的节点，所以默认它会取当前partition最新的Offset(Kafka自己维护的单个partition上递增序号)。

    理论上，如果找不到保存的Offset信息，应该从-1的Offset读起。

    这个问题我给作者提出来了，但作者认为这样可以避免重复处理，我没有想通为何会有重复处理。但好在作者说会在后续版本加入参数来控制。

    刚去看了下，似乎作者已经在提交 8b764cd fix掉了。有兴趣的可以去试下。我是自己本地改了他的代码。

    以上问题已修复并合并。
  • 相关阅读:
    Django使用redis
    Django中static media的简单配置
    套接字，TCP，UDP
    nginx常用配置
    使用systemctl管理nginx
    jumpserver 安装
    elasticsearch7.x集群安装（含head、bigdesk、kibana插件）
    codepush安装
    mysql优化后的主配置文件
    nginx优化、负载均衡、rewrite
  • 原文地址：https://www.cnblogs.com/cn-leodream/p/6497249.html
Copyright © 2011-2022 走看看