Storm相关笔记(包括Kafka和HBase）

一、Apache Kafka

1、了解Kafka

1.1、Kafka是什么？有什么用？

• 是什么？

  1） Apache Kafka 是一个消息队列（生产者消费者模式）

  2） Apache Kafka 目标：构建企业中统一的、高通量、低延时的消息平台。

  3） 大多的是消息队列（消息中间件）都是基于JMS标准实现的，Apache Kafka 类似于JMS的实现。

• 有什么用？

  1） 作为缓冲，来异构、解耦系统。

1.2、kafka的特性

• 消息持久化（Kafka 基于文件系统来存储和缓存消息）。

• 高吞吐量（Kafka 将数据写到磁盘，但是在底层采用了零拷贝技术，所以速度比较快）。

• 高扩展性（Kafka 依赖ZooKeeper来对集群进行协调管理，同时在机器扩展时无需将整个集群停机）。

• 多客户端支持（Kafka 核心模块用Scala 语言开发，但提供多种开发语言接入，包括Java，Python等）。

• 安全机制（支持代理与ZooKeeper 连接身份验证，客户端读、写权限认证）。

• 数据备份（Kafka 可以为每个主题指定副本数，对数据进行持久化备份）。

• 轻量级（Kafka 的实例是无状态的，同时集群本身几乎不需要生产者和消费者的状态信息）。

• 消息压缩（Kafka 支持Gzip, Snappy 、LZ4 这3 种压缩方式，把多条消息压缩成MessageSet）。

1.3、消息队列有什么用？

• 解耦，异构

2、Kafka的架构

• Kafka Cluster：由多个服务器组成。每个服务器单独的名字broker（掮客）。

• Kafka Producer：生产者、负责生产数据。

• Kafka consumer：消费者、负责消费数据。

• Kafka Topic: 主题，一类消息的名称。存储数据时将一类数据存放在某个topci下，消费数据也是消费一样。

• ZooKeeper：Kafka的元数据都是存放在zookeeper中。

3、操作Kafka集群的两种方式

3.1、使用控制台运行 Kafka

• 创建一个topic（主题）：

  bin/kafka-topics.sh --create --zookeeper zk01:2181 --replication-factor 1 --partitions 1 --topic order

• 启动一个生产者，用来生产数据 ：

  bin/kafka-console-producer.sh --broker-list kafka01:9092 --topic order

• 启动给一个消费者，消费数据：

  bin/kafka-console-consumer.sh --zookeeper zk01:2181 --from-beginning --topic order

3.2、使用JavaAPI操作Kafka

• JavaAPI操作Kafka所需要的依赖：

  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-clients</artifactId>
  <version>0.11.0.1</version>

• 生产者相关操作：

  //创建Properties配置参数对象，并设置参数
  Properties props = new Properties();
  props.put("bootstrap.servers", "node01:9092");
  props.put("acks", "all");
  //创建一个KafkaProducer，Kafka生产者对象
  KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(props);
      for (int i = 0; i < 1000; i++) {
          // 发送数据 ,需要一个producerRecord对象,最少参数 String topic, V value
          kafkaProducer.send(new ProducerRecord<String, String>("order", "订单信息！"+i));
          Thread.sleep(100);
      }  
  }

• 消费者相关操作：

  // 1、创建配置参数对象，并连接集群
  Properties props = new Properties();
  props.put("bootstrap.servers", "node01:9092");
  props.put("group.id", "test");
  props.put("enable.auto.commit", "true");
  props.put("auto.commit.interval.ms", "1000");
  props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
  props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
  //2、创建Kafka的消费者对象
  KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(props);
  //3、订阅一个主题，订阅主题需传入List格式
  kafkaConsumer.subscribe(Arrays.asList("order"));
  //4、使用死循环不停拉取数据
  while (true) {
      ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(100);
      for (ConsumerRecord<String, String> record : consumerRecords) {
          System.out.println("消费的数据为：" + record.value());
      }
  }

• topic相关操作：

  由于主题的元数据信息是注册在 ZooKeeper 相应节点之中，所以对主题的操作实质是对ZooKeeper中记录主题元数据信息相关路径的操作。Kafka将对ZooKeeper的相关操作封装成一 个ZkUtils 类，井封装了一个AdrninUtils类调用ZkClient类的相关方法以实现对 Kafka 元数据 的操作，包括对主题、代理、消费者等相关元数据的操作。对主题操作的相关API调用较简单，相应操作都是通过调用AdminUtils类的方法来完成的。

  • 创建topic（一般常用方法一）：

    方法一：
    //参数：zookeeper的地址，session超时时间，连接超时时间，是否启用zookeeper安全机制
    zkUtils = ZkUtils.apply("node01:2181", 30000, 3000, JaasUtils.isZkSecurityEnabled());
    方法二：
    //参数：zkUtils，topic名称，partition数量，副本数量，参数，机架感知模式
    AdminUtils.createTopic(zkUtils, topicName, 1, 1, new Properties(), AdminUtils.createTopic$default$6());

  • 删除topic：

    //参数：zkUtils，topic名称，partition数量，副本数量，参数，机架感知模式
    AdminUtils.deleteTopic(zkUtils, topicName);

  • 判断是否存在:

    AdminUtils.topicExists(zkUtils, topicName);

4、Apache Kafka的工作原理

4.1、分片与副本机制

• 分区：当数据量非常大的时候，一个服务器存放不了，就将数据分成两个或者多个部分，存放在多台服务器上。每个服务器上的数据，叫做一个分区。

• 副本：当数据只保存一份时，有丢失风险。为了更好的容错和容灾，将数据拷贝几份，保存到其他机器上。

• 设置分区和副本的方法：

  • 控制台上：--replication-factor 1 --partitions 3

  • API代码：AdminUtils.createTopic(zkUtils, topicName, 3, 1, new Properties(),

    AdminUtils.createTopic$default$6());

4.2、消息不丢失机制

• 生产者消息不丢失机制：

  • 发送消息的同步和异步模式：

    • 同步模式：生产者重试3次，如果还没有响应，就报错。生产者等待10S，如果broker没有给出ack响应，就认为失败。

    • 异步模式：先将数据保存在生产者端的buffer中。Buffer大小是2万条。发送一批数据的大小是500条。满足数据阈值或者数量阈值其中的一个条件就可以发送数据。

  • 消息确认的三个状态：

    • 0状态：生产者只负责发送数据，不管Kafka的broker是否接收到数据；

    • 1状态：某个partition的leader收到数据给出响应；

    • -1状态：某个partition的所有副本都收到数据后给出响应

• Borker端消息不丢失机制：

  broker端的消息不丢失，其实就是用partition副本机制来保证。Producer ack -1. 能够保证所有的副本都同步好了数据。其中一台机器挂了，并不影像数据的完整性。

• 消费者端消息不丢失：

  如果有一个外部存储能够记录每个consumer消费partition的offset值。就不会造成数据丢失，只会有重复消费的可能。而在Kafka0.8以后，offset值可以存放到Kafka内置的topic中。

4.3、消息存储和查询机制

4.3.1、消息存储机制

• Kafka作为消息中间件，只负责消息的临时存储，并不是永久存储，需要删除过期的数据；

• 如果一个partition中有10T数据，是如何存放的？是存放在一个文件中，还是存放在多个文件中？

  Kafka时采用存储到多个文件中的方式。因为如果将所有数据都存放在一个文件中，需要删除过期数据的时候，就比较麻烦。因为文件有日期属性，删除过期数据，只需要根据文件的日期属性删除就好。

• Kafka的数据是存储在/export/data/kafka(可以自己设置)目录下，存储时是将数据划分为一个个的segment段，在segment段中有两个核心的文件，一个是log，一个是index。当log文件等于1G时，新的会写入到下一个segment中。

4.3.2、消息查询机制

• 在Kafka中进行消息查询时，首先会查找segment中的index索引文件，index索引文件是以起始来命名的，根据查询索引文件能很快的定位到具体文件。

• 当根据index索引文件定位到需要查询的具体文件时，就会去查找log文件，在该文件中按顺序查找到目标文件

4.4、生产者数据分发策略

• kafka在数据生产的时候，有一个数据分发策略。默认的情况使用DefaultPartitioner.class类。如果用户制定了partition，生产就不会调用DefaultPartitioner.partition()方法。

• 当用户指定key，就会使用hash算法来确定发往那个patition。如果key一直不变，同一个key算出来的hash值是个固定值。如果是固定值，这种hash取模就没有意义。

  例：Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions

• 还可以指定将数据发往哪个partition。当ProducerRecord 的构造参数中有partition的时候，就可以发送到对应partition上。

  例：public ProducerRecord(String topic, Integer partition, K key, V value)

• 如果既没有指定partition，也没有key的情况下，那就使用轮询的方式发送数据。

4.5、消费者的负载均衡机制

一个partition只能被一个组中的成员消费。所以如果消费组中有多于partition数量的消费者，那么一定会有消费者无法消费数据。如果消费组中的消费组小于partition，那么消费的数据就不完整，会造成错误。

5、Spring-Kafka

5.1、Spring-Kafka的概述和依赖

• 概述：Spring对kafka做了支持，以便简化我们的开发工作，官网：https://spring.io/projects/spring-kafka

• 依赖：

  <dependency>
      <groupId>org.springframework.kafka</groupId>
      <artifactId>spring-kafka</artifactId>
      <version>2.1.7.RELEASE</version>
  </dependency>
  <dependency>
      <groupId>org.springframework</groupId>
      <artifactId>spring-test</artifactId>
      <version>5.0.6.RELEASE</version>
  </dependency>    

5.2、Spring-Kafka生产者示例代码

第一步：编写application-kafka-producer.xml文件

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:context="http://www.springframework.org/schema/context"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
        http://www.springframework.org/schema/beans/spring-beans.xsd
        http://www.springframework.org/schema/context
        http://www.springframework.org/schema/context/spring-context.xsd">
​
    <bean id="producerConfig" class="java.util.HashMap">
        <constructor-arg>
            <map>
                <entry key="bootstrap.servers" value="node01:9092,node02:9092"/>
                <!--设置json序列化-->
                <entry key="key.serializer" value="org.springframework.kafka.support.serializer.JsonSerializer"/>
                <entry key="value.serializer" value="org.springframework.kafka.support.serializer.JsonSerializer"/>
            </map>
        </constructor-arg>
    </bean>
​
    <!-- 定义生产者的工厂 -->
    <bean id="producerFactory" class="org.springframework.kafka.core.DefaultKafkaProducerFactory">
        <constructor-arg ref="producerConfig"/>
    </bean>
​
    <!-- 定义KafkaTemplate,通过KafkaTemplate即可实现发送消息等功能 -->
    <bean class="org.springframework.kafka.core.KafkaTemplate">
        <constructor-arg index="0" ref="producerFactory"/>
        <!--指定默认的topic-->
        <property name="defaultTopic" value="my-kafka-topic"/>
    </bean>
​
</beans>

第二步：编写java代码 TestSpringKafkaProducer

//类上方添加如下注释
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(value = {"classpath:application-kafka-producer.xml"})
//注入一个KafkaTemplate对象
@Autowired
private KafkaTemplate kafkaTemplate;
//使用注入的对象发送数据到Kafka（发送的数据可以是对象，会自动进行json转换）
kafkaTemplate.sendDefault(order);

5.3、Spring-Kafka消费者示例代码

第一步：编写application-kafka-consumer.xml文件

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:context="http://www.springframework.org/schema/context"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
        http://www.springframework.org/schema/beans/spring-beans.xsd
        http://www.springframework.org/schema/context
        http://www.springframework.org/schema/context/spring-context.xsd">

    <bean id="consumerConfig" class="java.util.HashMap">
        <constructor-arg>
            <map>
                <entry key="bootstrap.servers" value="node01:9092,node02:9092"/>
                <entry key="group.id" value="my-group-spring-spring-3"/>
                <entry key="client.id" value="my-test-client-spring-3"/>
                <entry key="enable.auto.commit" value="true"/>
                <entry key="auto.commit.interval.ms" value="1000"/>
                <!--反序列化器，这里要注意设置的是字符串的反序列化-->
                <entry key="key.deserializer" value="org.apache.kafka.common.serialization.StringDeserializer"/>
                <entry key="value.deserializer" value="org.apache.kafka.common.serialization.StringDeserializer"/>
            </map>
        </constructor-arg>
    </bean>

    <!-- 定义消费者的工厂 -->
    <bean id="consumerFactory" class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">
        <constructor-arg ref="consumerConfig"/>
    </bean>

    <!--定义消息监听器，用于接收消息-->
    <bean id="myMessageListener" class="cn.itcast.kafka.MyMessageListener"/>

    <bean id="containerProperties" class="org.springframework.kafka.listener.config.ContainerProperties">
        <!--设置消费的topic，这里可以指定多个topic-->
        <constructor-arg value="my-kafka-topic" type="java.lang.String[]"/>
        <property name="messageListener" ref="myMessageListener"/>
    </bean>

    <!--创建Listener容器-->
        <bean class="org.springframework.kafka.listener.KafkaMessageListenerContainer" init-method="start">
        <constructor-arg index="0" ref="consumerFactory"/>
        <constructor-arg index="1" ref="containerProperties"/>
    </bean>

</beans>

第二步：创建一个类，注入上述配置文件即可接收

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(value = {"classpath:application-kafka-consumer.xml"})
public class TestSpringKafkaConsumer {
    @Test
    public void testConsumer() {
    }
}

6、Kafka和Kafka-manager

• Kafka Manager 由 yahoo 公司开发，该工具可以方便查看集群 主题分布情况，同时支持对 多个集群的管理、分区平衡以及创建主题等操作。

• Kafka和Kafka-manager的详细安装和部署，请详看其安装部署文件。

• 启动Kafka-manager：

  cd /export/servers/kafka-manager-1.3.3.17/bin ./kafka-manager -Dconfig.file=../conf/application.conf

二、Storm

1、什么是Storm

• 官网：http://storm.apache.org/ 源码：https://github.com/apache/storm

• Storm是一个开源免费的分布式实时计算系统，Storm可以轻松的处理无界的数据流。

• Storm只负责数据的计算，不负责数据的存储。

• 2013年前后，阿里巴巴基于storm框架，使用java语言开发了类似的流式计算框架佳作，Jstorm。2016年年底阿里巴巴将源码贡献给了Apache storm，两个项目开始合并，新的项目名字叫做storm2.x。阿里巴巴团队专注flink开发。

2、流式计算的架构

3、Strom的架构

3.1、集群架构

• Nimbus：负责资源分配和任务调度。
• Supervisor：负责接受nimbus分配的任务，启动和停止属于自己管理的worker进程。
• Worker：运行具体处理组件逻辑的进程。
• Task：worker中每一个spout/bolt的线程称为一个task. 在storm0.8之后，task不再与物理线程对应，同一个spout/bolt的task可能会共享一个物理线程，该线程称为executor。

架构说明：

1. 在集群架构中，用户提交到任务到storm，交由nimbus处理。

2. nimbus通过zookeeper进行查找supervisor的情况，然后选择supervisor进行执行任务。

3. supervisor会启动一个woker进程，在worker进程中启动线程进行执行具体的业务逻辑。

3.2、编程模型

• Spout：Spout继承BaseRichSpout，其中有三个方法：open（用来进行初始化），nextTuple（storm框架会不断调用该方法进行执行，向下游发送数据），declareOutputFields（定义向下游发送的数据的名称，定义的名称和发送的数据的顺序要一致）。

• Bolt：Bolt继承BaseRichBolt，其中有三个方法：prepare（初始化操作，只会执行一次），execute（storm框架会不断调用该方法进行执行，处理业务逻辑，向下游发送数据），declareOutputFields（定义向下游发送的数据的名称，定义的名称和发送的数据的顺序要一致）。

• Tuple：一次消息传递的基本单元。本来应该是一个key-value的map，但是由于各个组件间传递的tuple的字段名称已经事先定义好，所以tuple中只要按序填入各个value就行了，所以就是一个value list.

• Stream：源源不断传递的tuple就组成了stream。

• Topology：Storm中运行的一个实时应用程序，因为各个组件间的消息流动形成逻辑上的一个拓扑结构。

  创建Topology的步骤：

  第一步，创建一个TopologyBuilder；

  第二步，向TopologyBuilder中设置Spout和Bolt，并且进行连接；

  第三步，topologyBuilder.createTopology()得到Topology对象；

  第四步，提交到Storm进行运行（本地模式：localCluster.submitTopology()，集群模式：

  StormSubmitter.submitTopology()）；

4、优化提交Topology的逻辑

4.1、为什么要优化提交Topology的逻辑

因为频繁切换本地和集群模式太麻烦了，当进行Topology优化后，本地和集群就可以相同代码，当启动时，可以根据main方法中args做出判断，如果args没有参数，说明是本地模式，如果有参数，说明是集群模式。

4.2、优化Topology的好处

优化之后，本地模式和集群模式可以使用相同的代码，不要进行切换，而且在集群模式下，可以对Topology进行自定义名称。

5、Storm的容错机制

5.1、worker宕机怎么办？

• 由于worker和supervisor分离，当worker宕机，但supervisor能运行时，supervisor会尝试重新启动worker

• 如果supervisor也宕机了，那nimbus会重新分配其它的supervisor进行执行

5.2、supervisor宕机怎么办？

• 如果只有supervisor宕机，但其下面的worker没有宕机时，那worker会正常工作，不会有影响

• 如果supervisor及其下面的worker宕机，首先会将分配给该机器的任务暂停，并且nimbus会重新分配机器来执行该节点上的任务

• 所以需要借助外部的监控手段来保障supervisor的高可用

5.3、nimbus宕机怎么办？

• 当nimbus宕机时，不会影响任务的执行，将影响的是，任务的提交，不能再向集群提交任何任务

• 所以需要配置nimbus的高可用：可以配置、启动多个nimbus来保障；也可以需要借助外部的监控手段来保障nimbus的高可用

6、Storm与其他框架的整合

6.1、storm整合jdbc

有些时候我们需要将Storm计算完的数据持久化到数据库，所以需要在Storm中整合JDBC进行持久化。

官方文档：https://github.com/apache/storm/blob/master/docs/storm-jdbc.md

• 引入storm-jdbc和mysql驱动依赖

      <dependency>
          <groupId>mysql</groupId>
          <artifactId>mysql-connector-java</artifactId>
          <version>5.1.34</version>
      </dependency>
      <dependency>
          <groupId>org.apache.storm</groupId>
          <artifactId>storm-jdbc</artifactId>
          <version>1.1.1</version>
      </dependency>
  

• 使用JdbcInsertBolt进行整合的步骤

  public static IRichBolt build() {
      //第一步：定义数据库连接信息，包括数据库驱动和数据库地址，用户名，密码等
      Map<String, Object> hikariConfigMap = new HashMap<String, Object>();
   hikariConfigMap.put("dataSourceClassName","com.mysql.jdbc.jdbc2.optional.MysqlDataSource");
      hikariConfigMap.put("dataSource.url", "jdbc:mysql://localhost:3306/storm");
      hikariConfigMap.put("dataSource.user", "root");
      hikariConfigMap.put("dataSource.password", "root");
      ConnectionProvider connectionProvider = new HikariCPConnectionProvider(hikariConfigMap);
  
      // 第二步：定义表名，以及定义字段的映射，这里指定的是tupe中的字段名称，用于获取数据
      String tableName = "tb_wordcount";
      List<Column> columnSchema = Lists.newArrayList(
          new Column("word", Types.VARCHAR),
          new Column("count", Types.INTEGER));
  
      // 第三步：定义jdbc的映射器
      JdbcMapper simpleJdbcMapper = new SimpleJdbcMapper(columnSchema);
  
      //第四步： 定义插入数据的Bolt，并且指定了插入的sql语句
      JdbcInsertBolt wordCountBolt = new JdbcInsertBolt(connectionProvider, simpleJdbcMapper)
          .withInsertQuery("INSERT INTO `tb_wordcount` VALUES (NULL, ?, ?, NOW())")
          .withQueryTimeoutSecs(30);
  
      //第五步：将结果进行返回
      return wordCountBolt;
  }
  

  //整合到TopologyBuilder中
  topologyBuilder.setBolt("JdbcBolt",JdbcBoltBuilder.build()).shuffleGrouping("WordBolt");
  

6.2、Storm与Redis整合

Storm和Redis整合是非常常用的场景，Storm也支持了Redis的支持。

官方文档：https://github.com/apache/storm/blob/master/docs/storm-redis.md

• 引入依赖

  <dependency>
      <groupId>org.apache.storm</groupId>
      <artifactId>storm-redis</artifactId>
      <version>1.1.1</version>
  </dependency>
  

• 创建WordCountStoreMapper

  public class WordCountStoreMapper implements RedisStoreMapper {
      private RedisDataTypeDescription redisDataTypeDescription;
      public WordCountStoreMapper() {
          // 定义Redis中的数据类型
          this.redisDataTypeDescription =
                  new RedisDataTypeDescription(RedisDataTypeDescription.RedisDataType.STRING);
      }
      
      @Override
      public RedisDataTypeDescription getDataTypeDescription() {
          return this.redisDataTypeDescription;
      }
  
      @Override
      public String getKeyFromTuple(ITuple iTuple) {
          // 生成redis中的key
          String word = iTuple.getStringByField("word");
          return "wordCount:" + word;
      }
  
      @Override
      public String getValueFromTuple(ITuple iTuple) {
          // 存储到redis中的值
          Integer count = iTuple.getIntegerByField("count");
          return String.valueOf(count);
      }
  }
  

• 整合到Topology中

  //创建Redis的连接参数
  JedisPoolConfig poolConfig = new JedisPoolConfig.Builder() 
      										.setHost("node01").setPort(6379).build();
  //根据Redis的连接参数和Redis的Mapper类，创建Redis的RedisStoreBolt对象，并将该对象整合到Topology中
  topologyBuilder.setBolt("RedistBolt", new RedisStoreBolt(poolConfig, new 			
      					 WordCountStoreMapper())).localOrShuffleGrouping("WordCountBolt");
  

6.3、Storm与Kafka整合

Storm与Kafka的整合也是非常常见的，常常用于数据读取，所以我们更多的是要关注Spout。

官方文档：https://github.com/apache/storm/blob/master/docs/storm-kafka-client.md

• 引入依赖

  <dependency>
      <groupId>org.apache.storm</groupId>
      <artifactId>storm-kafka-client</artifactId>
      <version>1.1.1</version>
  </dependency>
  

• 整合KafkaSpout到Topology

  //已知存在kafka的一个topic，该topic有3个分区，2个副本
  
  //第一步，定义TopologyBuilder对象，用于构建拓扑
  TopologyBuilder topologyBuilder = new TopologyBuilder();
  
  //第二步，设置spout和bolt，创建kafka的Spout，并连接到kafka的模板topic上，并设置消费者id
  KafkaSpoutConfig.Builder<String, String> kafkaSpoutBuilder = 		
      						KafkaSpoutConfig.builder("node01:9092", "kafka-storm-topic");
  kafkaSpoutBuilder.setGroupId("kafka-storm-topic-consumer-groupid"); //设置消费者组id
  
  // 这里设置Spout的并行度为3，原因是创建topic时，指定的partition为3
  topologyBuilder.setSpout("kafka_spout", new KafkaSpout<>(kafkaSpoutBuilder.build()),3);
  //设置一个bolt，并指定从kafka中获取数据的spout为这个bolt的上游
  topologyBuilder.setBolt("SplitSentenceBolt", new 
                          		SplitSentenceBolt()).localOrShuffleGrouping("kafka_spout");
  

• 当上游为Kafka的Spout时，下游Bolt获取值的方法

  通过KafkaSpout向下游发送的Tupe的方式是这样的：collect.emti();

  所以，我们需要通过value获取值：String sentence = input.getStringByField("value");

7、Strom的流分组

7.1、为什么要有流分组

在Storm中，当Spout、Bolt是多个时，那上游的Spout或Bolt不知道将数据传入下游的哪个Bolt时，就需要流分组

7.2、常用的分组方式

在Storm中，提供了8种流分组方式，其中3中经常使用

• 随机分组（Shuffle grouping）

  随机分发tuple到Bolt的任务，保证每个任务获得相等数量的tuple。 跨服务器通信，浪费网络资源，尽量不用

• 本地或随机分组（Local or shuffle grouping）（重点掌握，常用）

  优先将数据发送到本地的Task，节约网络通信的资源

• 部分关键字分组（Partial Key grouping）

• 字段分组（Fields Grouping ）（了解，仅用于WordCount案例）

8、Storm的通信机制

8.1、什么是通信机制

在不同的worker间进行数据传输时，会产生通信（相同主机的进程通信，不同主机需通过网络通信）

8.2、Worker 内部通信技术(Disruptor)

Disruptor是一个Queue。Disruptor是实现了“队列”的功能，而且是一个有界队列。而队列的应用场景自然就是“生产者-消费者”模型。

Disruptor的核心有3个：第一，维护生产者序号；第二，消费者序号；第三，数组的长度。

8.3、Disruptor 为什么这么快

• 原因一：Disruptor 没有使用锁机制；

• 原因二：Disruptor 有一个类似中间数组的组件，生产者往这个组件中放入数据，消费者取出数据

9、Storm的消息不丢失机制

9.1、原理

Storm的消息不丢失机制核心是acker。当Spout或Bolt处理完数据后，会标记状态到acker中，然后通过异或算法，计算出是否有失败的情况。结果为0就表示执行成功，结果不为0就表示失败。

当结果为0时，acker会通知Spout执行成功，从而调用Spout的ack()方法，此时可以在ack()方法中执行操作。

当结果不为0时，acker会通知Spout失败，从而调用它的fail()方法，此时可以在该方法中重新执行。

9.2、实现

9.2.1、复杂实现，了解即可

• Spout实现：在Spout中需实现ack和fail方法，并且在Spout中发送消息时需要发送消息id（唯一，实现如下）

  // 生成消息id，并且把数据存放到messages的map中，如果执行失败可以在fail方法中调用messages中的数据重新执行
  String msgId = UUID.randomUUID().toString();
  messages.put(msgId, sentence);
  //向下游输出
  this.collector.emit(new Values(sentence),msgId);
  

• Bolt中的实现：如果执行成功就调用ack方法，如果执行失败就调用fail方法

  public void execute(Tuple input) {
      // 通过Tuple的getValueByField获取上游传递的数据，其中"sentence"是定义的字段名称
      String sentence = input.getStringByField("sentence");
      
      // 将获取的数据进行处理
  
      // 向下游输出数据，需加入锚点
      for (String word : words) {
          // 注意这里，需要将原始数据input传入，在这里称之为锚点，意思是将新的数据和原有数据进行关联
          this.collector.emit(input,new Values(word)); 
      }
  
      // 如果处理成功，执行ack方法，如果处理失败，执行fail方法，并将传入的input传入该方法中
      this.collector.ack(input);
      this.collector.fail(input);
  }
  

9.2.2、简单实现，需重点掌握

从上看出，如果执行失败需手动调用方法，很不方便，所以我们可以继承BaseBasicBolt类，这样当执行失败时就不需要手动调用方法了，只需要抛出FailedException即可。请注意：就是继承BaseBasicBolt类，在Spout中还是需重写ack和fail方法，这是当执行成功或失败后会调用的方法。

• 好处一：如果执行失败，不需要手动调用fail和ack方法，只需要使用try...catch包裹执行代码，然后抛出FailedException异常即可。

• 好处二：向下游输出数据时，不需要加入锚点，只需要将目标数据加入。

• 好处三：可以在Topology驱动类中设置acker（消息不丢失机制）是否启动，设置方法如下

  Config.setNumAckers(conf, ackerParal);

9.3、消息不丢失机制的处理逻辑

首先在Spout中需将发送到下游的数据进行记录（可以记录到一个Map中）。如果该数据执行成功，可以将Map中的该数据删除（this.msgData.remove(msgId);）。如果该数据执行失败，可以在fail方法中重新执行该数据，就是将Map中对应的数据取出，然后重新发送到下游执行。

如果执行失败，最好设置重新执行的限制条件：需限制失败的重试次数，如果重试次数超出，需将该数据记录下来（可以保存到数据库等本地文件中），同时删除内存中的数据（Map集合中的该数据）；在进行重试时，需要有停顿，不能失败后马上执行（有可能是网络问题），需要有计数，不能无限执行。

三、Mahout协同过滤算法

1、推荐系统

1.1、什么是推荐系统

为了解决信息过载和用户无明确需求的问题，找到用户感兴趣的物品，才有了个性化推荐系统。

1.2、推荐系统业务流程

推荐系统广泛存在于各类网站中，作为一个应用为用户提供个性化的推荐。它需要一些用户的历史数据，一般由三个部分组成：基础数据、推荐算法系统、前台展示。

• 基础数据包括很多维度，包括用户的访问、浏览、下单、收藏，用户的历史订单信息，评价信息等很多信息；

• 推荐算法系统主要是根据不同的推荐诉求由多个算法组成的推荐模型；

• 前台展示主要是对客户端系统进行响应，返回相关的推荐信息以供展示。

2、协同过滤算法

2.1、协同过滤算法概述

迄今为止，在个性化推荐系统中，协同过滤技术是应用最成功的技术。目前国内外有许多大型网站应用这项技术为用户更加智能（个性化、千人千面）的推荐内容。

• 核心思想：协同过滤一般是在海量的用户中发掘出一小部分和你品位比较类似的，在协同过滤中，这些用户成为邻居，然后根据他们喜欢的其他东西组织成一个排序的目彔作为推荐给你。

• 问题：如何确定一个用户是丌是和你有相似的品位？如何将邻居们的喜好组织成一个排序的目彔？

2.2、如何选择？

• 在社交项目中，如微信、QQ，显然选择基于用户推荐比较好，因为推荐往往都是和人相关的。

  • 如：在QQ登录后，会有提示，好友的好友可能是你认识的，推荐给你添加好友。

• 在电商项目中，用户的数量远大于商品数量，所以基于商品的推荐的复杂度要低，而且也比较合理。

• 其实，在实际的推荐系统中，往往不单是使用一种推荐，而是会多种推荐混合使用。

• 所以，选择基于用户还是基于商品的推荐，和应用场景有很大的关系。

3、相似度算法

无论是基于用户还是基于商品的推荐，都是需要找到相似的用户或者商品，才能做推荐，所以，相似度算法就变得非常重要了。

常见的相似度算法有：

• 欧几里德距离算法（Euclidean Distance）

• 皮尔逊相似度算法（Pearson Correlation Coefficient）

• 基于夹角余弦相似度算法（Consine Similarity）

• 基于Tanimoto系数相似度（Tanimoto Coefficient）

4、最近邻域

通过相似度计算，可以计算出邻居，问题来了，我们如果选取出几个邻居作为参考，进行推荐呢？

通常有2种方式：

• 固定数量的邻居：K-neighborhoods

• 基于相似度门槛的邻居：Threshold-based neighborhoods

5、Mahout的概述

Mahout使用了Taste来提高协同过滤算法的实现，它是一个基于Java实现的可扩展的，高效的推荐引擎。Taste既实现了最基本的基于用户的和基于内容的推荐算法，同时也提供了扩展接口，使用户可以方便的定义和实现自己的推荐算法。同时，Taste不仅仅只适用于Java应用程序，它可以作为内部服务器的一个组件以HTTP和Web Service的形式向外界提供推荐的逻辑。Taste的设计使它能满足企业对推荐引擎在性能、灵活性和可扩展性等方面的要求。

5.1、Mahout的起源

• 2008年成为Lucene的子项目，Lucene作为搜索引擎项目，存在很多文本数据分析和挖掘的需求（例如文本重复判断，文本自动分类等等），导致Lucene项目中部分开发者转向机器学习领域研究算法，最终这些机器学习算法形成最初的Mahout

• 吸收开源协同过滤算法项目Taste

• 2010年成为Apache顶级项目

5.2、Mahout的能做什么

• Mahout提供了基于用户和基于商品的推荐系统的协同过滤算法；
• 提供常用的相似度算法，如：欧几里得距离算法、皮尔逊相似度算法等；
• Mahout提供了2种最近邻域的实现；
• Mahout提供了多种数据源实现，可以读取文件、数据库、Hbase等；

6、Taste 中的主要接口

6.1、DataModel（数据模块）接口

DataModel 是用户喜好信息的抽象接口，它的具体实现支持从任意类型的数据源抽取用户喜好信息。Taste 默认提供 JDBCDataModel 和 FileDataModel，分别支持从数据库和文件中读取用户的喜好信息。

DataModel接口的部分实现：
	org.apache.mahout.cf.taste.impl.model.GenericDataModel
	org.apache.mahout.cf.taste.impl.model.GenericBooleanPrefDataModel
	org.apache.mahout.cf.taste.impl.model.PlusAnonymousUserDataModel
	org.apache.mahout.cf.taste.impl.model.file.FileDataModel
	org.apache.mahout.cf.taste.impl.model.hbase.HBaseDataModel

6.2、UserSimilarity 和 ItemSimilarity（相似度） 接口

UserSimilarity 和 ItemSimilarity 。UserSimilarity 用于定义两个用户间的相似度，它是基于协同过滤的推荐引擎的核心部分，可以用来计算用户的“邻居”，这里我们将与当前用户口味相似的用户称为他的邻居。ItemSimilarity 类似的，计算Item之间的相似度。

UserSimilarity 和 ItemSimilarity 相似度实现有以下几种：
	CityBlockSimilarity ：基于Manhattan距离相似度
	EuclideanDistanceSimilarity ：基于欧几里德距离计算相似度
	LogLikelihoodSimilarity ：基于对数似然比的相似度
	PearsonCorrelationSimilarity ：基于皮尔逊相关系数计算相似度
	SpearmanCorrelationSimilarity ：基于皮尔斯曼相关系数相似度
	TanimotoCoefficientSimilarity ：基于谷本系数计算相似度
	UncenteredCosineSimilarity ：计算 Cosine 相似度

6.3、UserNeighborhood（最近邻域）接口

UserNeighborhood 用于基于用户相似度的推荐方法中，推荐的内容是基于找到与当前用户喜好相似的邻居用户的方式产生的。UserNeighborhood 定义了确定邻居用户的方法，具体实现一般是基于 UserSimilarity 计算得到的。

UserNeighborhood 主要实现有两种：
	NearestNUserNeighborhood：对每个用户取固定数量N个最近邻居
	ThresholdUserNeighborhood：对每个用户基于一定的限制，取落在相似度限制以内的所有用户为邻居

6.4、Recommender（推荐引擎）接口

Recommender 是推荐引擎的抽象接口，Taste 中的核心组件。程序中，为它提供一个 DataModel，它可以计算出对不同用户的推荐内容。实际应用中，主要使用它的实现类 GenericUserBasedRecommender 或者 GenericItemBasedRecommender，分别实现基于用户相似度的推荐引擎或者基于内容的推荐引擎。

Recommender分为以下几种实现：
	GenericUserBasedRecommender：基于用户的推荐引擎
	GenericBooleanPrefUserBasedRecommender：基于用户的无偏好值推荐引擎
	GenericItemBasedRecommender：基于物品的推荐引擎
	GenericBooleanPrefItemBasedRecommender：基于物品的无偏好值推荐引擎

6.5、RecommenderEvaluator（推荐系统评测）接口

RecommenderEvaluator有以下几种实现：
	AverageAbsoluteDifferenceRecommenderEvaluator ：计算平均差值
	RMSRecommenderEvaluator ：计算均方根差

7、Mahout的实现

7.1、实现基于用户的推荐 UserCF

 @Test
    public void testBaseUser() throws Exception {
        String fileName = "user_item.data";
        File file = FileUtils.toFile(TestMahout.class.getClassLoader().getResource(fileName));
        // 第一步，定义数据模型
        DataModel dataModel = new FileDataModel(file);

        // 第二步，定义相识度，这里使用的欧几里得
        UserSimilarity userSimilarity = new EuclideanDistanceSimilarity(dataModel);

        // 第三步，定义最近邻域，这里使用的是固定数量的邻居
        UserNeighborhood userNeighborhood = new NearestNUserNeighborhood(10, userSimilarity, dataModel);
        long[] longs = userNeighborhood.getUserNeighborhood(1);
        for (long aLong : longs) {
            System.out.println(aLong);
        }
        // 第四步，定义推荐器，这里使用的是基于用户的推荐
        Recommender recommender = new GenericUserBasedRecommender(dataModel, userNeighborhood, userSimilarity);

        LongPrimitiveIterator userIDs = dataModel.getUserIDs();
        while (userIDs.hasNext()) {
            Long userId = userIDs.next();
            List<RecommendedItem> recommendedItemList = recommender.recommend(userId, 4);

            StringBuffer sb = new StringBuffer();
            for (RecommendedItem item : recommendedItemList) {
                sb.append(item.getItemID() + "|"+item.getValue()+",");
            }

            System.out.println(userId + "-->" + sb);
        }

    }

7.2、实现基于商品的推荐

@Test
    public void testBaseItem() throws Exception {
        String fileName = "user_item.data";
        File file = FileUtils.toFile(TestMahout.class.getClassLoader().getResource(fileName));
        // 第一步，定义数据模型
        DataModel dataModel = new FileDataModel(file);

        // 第二步，定义相识度，这里使用的欧几里得
        ItemSimilarity itemSimilarity = new EuclideanDistanceSimilarity(dataModel);

        // 第三步，定义推荐器，这里使用的是基于用户的推荐
        Recommender recommender = new GenericItemBasedRecommender(dataModel, itemSimilarity);

        LongPrimitiveIterator userIDs = dataModel.getUserIDs();
        while (userIDs.hasNext()) {
            Long userId = userIDs.next();
            List<RecommendedItem> recommendedItemList = recommender.recommend(userId, 2);

            StringBuffer sb = new StringBuffer();
            for (RecommendedItem item : recommendedItemList) {
                sb.append(item.getItemID() + "|"+item.getValue()+",");
            }

            System.out.println(userId + "-->" + sb);
        }

    }

7.3、根据用户浏览记录推荐

@Test
    public void testBaseItem2() throws Exception {
        String fileName = "user_item.data";
        File file = FileUtils.toFile(TestMahout.class.getClassLoader().getResource(fileName));
        // 第一步，定义数据模型
        DataModel dataModel = new FileDataModel(file);

        // 第二步，定义相识度，这里使用的欧几里得
        ItemSimilarity itemSimilarity = new EuclideanDistanceSimilarity(dataModel);

        // 第三步，定义推荐器，这里使用的是基于用户的推荐
        GenericItemBasedRecommender recommender = new GenericItemBasedRecommender(dataModel, itemSimilarity);

        LongPrimitiveIterator userIDs = dataModel.getUserIDs();
        //用户1，正在浏览103商品，进行推荐
        Long userId = 1L;
        Long itemId = 103L;
        List<RecommendedItem> recommendedItemList = recommender.recommendedBecause(userId,itemId,2);

        StringBuffer sb = new StringBuffer();
        for (RecommendedItem item : recommendedItemList) {
            sb.append(item.getItemID() + "|"+item.getValue()+",");
        }

        System.out.println(userId + "-->" + sb);

    }

8、将Mahout提交到hadoop运行

提交到hadoop运行，需要有2步操作：

第一步，将需要计算的数据上传到hdfs

第二步，通过hadoop执行mahout-examples-0.13.0-job.jar中的RecommenderJob类

第三步，在输出结果文件中查看结果

hadoop jar mahout-examples-0.13.0-job.jar org.apache.mahout.cf.taste.hadoop.item.RecommenderJob --input /user_item.data --output /cc -s SIMILARITY_EUCLIDEAN_DISTANCE

参数说明：

--input(path)(-i): 存储用户偏好数据的目录，该目录下可以包含一个或多个存储用户偏好数据的文本文件；
--output(path)(-o): 结算结果的输出目录
--numRecommendations (integer): 为每个用户推荐的item数量，默认为10
--usersFile (path): 指定一个包含了一个或多个存储userID的文件路径，仅为该路径下所有文件包含的userID做推荐计算 (该选项可选)
--itemsFile (path): 指定一个包含了一个或多个存储itemID的文件路径，仅为该路径下所有文件包含的itemID做推荐计算 (该选项可选)
--filterFile (path): 指定一个路径，该路径下的文件包含了[userID,itemID]值对，userID和itemID用逗号分隔。计算结果将不会为user推荐[userID,itemID]值对中包含的item (该选项可选)
--booleanData (boolean): 如果输入数据不包含偏好数值，则将该参数设置为true，默认为false
--maxPrefsPerUser (integer): 在最后计算推荐结果的阶段，针对每一个user使用的偏好数据的最大数量，默认为10
--minPrefsPerUser (integer): 在相似度计算中，忽略所有偏好数据量少于该值的用户，默认为1
--maxSimilaritiesPerItem (integer): 针对每个item的相似度最大值，默认为100
--maxPrefsPerUserInItemSimilarity (integer): 在item相似度计算阶段，针对每个用户考虑的偏好数据最大数量，默认为1000
--similarityClassname (classname)（-s）: 向量相似度计算类
([SIMILARITY_COOCCURRENCE,    
                                                  SIMILARITY_LOGLIKELIHOOD,     
                                                  SIMILARITY_TANIMOTO_COEFFICIEN
                                                  T, SIMILARITY_CITY_BLOCK,     
                                                  SIMILARITY_COSINE,            
                                                  SIMILARITY_PEARSON_CORRELATION
                                                  ,                             
                                                  SIMILARITY_EUCLIDEAN_DISTANCE]
                                                  )
outputPathForSimilarityMatrix：SimilarityMatrix输出目录
--randomSeed：随机种子 –sequencefileOutput：序列文件输出路径
--tempDir (path): 存储临时文件的目录，默认为当前用户的home目录下的temp目录
--startPhase
--endPhase
--threshold (double): 忽略相似度低于该阀值的item对

四、HBase

1、了解HBase

1.1、什么是HBase？

官网：https://hbase.apache.org/

补充网站：https://blog.csdn.net/u013054888/article/details/53510849

HBase是一个使用Java语言实现的，构建于Hadoop分布式文件系统(HDFS)上的分布式数据库。

Hbase是参考谷歌的BigTable的论文开发实现的，Hadoop 生态系统引入了Bigtable的大部分功能。

1.2、HBase的特点

• 海量存储：Hbase单表可以有百亿行，百万列，相对计较传统关系型数据库而言，存储能力非常强悍。

• 列式存储：创建表时，无需指定具体的列，根据数据的插入动态插入；可以针对列进行权限控制和读取。

• 多版本：可以为数据添加版本信息，如用户信息的logo变更历史。

• 稀疏性：为空的列不占用实际存储空间。

• 高扩展、高可用性：底层基于HDFS，高可用和扩展性得到的了保障。

1.3、Hbase中表结构模型

 

• 表(table)：用于存储管理数据，具有稀疏的、面向列的特点。HBase中的每一张表，就是所谓的大表(Bigtable)。

• 行键(RowKey)：类似于MySQL中的主键，HBase根据行键来快速检索数据，一个行键对应一条记录。与MySQL主键不同的是，HBase的行键是天然固有的，每一行数据都存在行键。

• 列族（簇）(ColumnFamily)：是列的集合。列族在表定义时需要指定，而列在插入数据时动态指定。列中的数据都是以二进制形式存在，没有数据类型。在物理存储结构上，每个表中的每个列族单独以一个文件存储(参见图1.2)。一个表可以有多个列族。

• 时间戳(TimeStamp)：是列的一个属性，是一个64位整数。由行键和列确定的单元格，可以存储多个数据，每个数据含有时间戳属性，数据具有版本特性。可根据版本(VERSIONS)或时间戳来指定查询历史版本数据，如果都不指定，则默认返回最新版本的数据。

2、Hbase系统架构

全局架构：

有此可以看出，Hbase需要依赖于ZooKeeper和HDFS。

• Zookeeper

  • 保证任何时候，集群中只有一个running master，避免单点问题；

  • 存贮所有Region的寻址入口，包括-ROOT-表地址、HMaster地址；

  • 实时监控Region Server的状态，将Region server的上线和下线信息，实时通知给Master；

  • 存储Hbase的schema，包括有哪些table，每个table有哪些column family。

• Master

  • 可以启动多个HMaster，通过Zookeeper的Master Election机制保证总有一个Master运行。

• RegionServer

  • HBase中最核心的模块，主要负责响应用户I/O请求，向HDFS文件系统中读写数据。

    • 维护Master分配给它的region，处理对这些region的IO请求；

    • 负责切分在运行过程中变得过大的region。

• HDFS

  • 负责存储数据。

3、Hbase控制台的CRUD操作

通过hbase shell命令进行命令行模式进行操作。

 

3.1、创建表

#指定表名，列族名
create 'user' , 'user_info', 'login_info'	#创建表 -表名 -列族1名 -列族2名
list									 #查看所有的表
describe 'user'							  #查看该表的具体信息

3.2、插入数据

put 'user', '1001', 'user_info:name','张三'
put 'user', '1001', 'user_info:address', '上海'
put 'user', '1001', 'login_info:user_name', 'zhangsan'
put 'user', '1001', 'login_info:password', '123456'

put 'user', '1002', 'user_info:name','李四'
put 'user', '1002', 'user_info:address', '北京'
put 'user', '1002', 'login_info:user_name', 'lisi'
put 'user', '1002', 'login_info:password', '123456'

3.3、查询数据

Hbase只支持2种查询数据，单行查询，全表查询。

get 'user', '1001'
#查询全部数据
scan 'user'
#查询一条数据
scan 'user', {LIMIT => 1}

3.4、删除数据

#删除一行中的一列数据
delete 'user','1002', 'user_info:name'
#删除一行数据
deleteall 'user','1002'
#清空表
truncate 'user'

3.5、修改数据

#修改用1001的密码为888888，直接put覆盖即可
put 'user', '1001', 'login_info:password', '888888'

#删除列族
alter 'user' , {NAME=>'user_info', METHOD => 'delete'}

#增加列族
alter 'user', 'user_info', {NAME => 'user_info_2' , VERSIONS => 5}

3.6、删除表

#删除表之前先要禁用表，再删除
disable 'user'
drop 'user'