电商推荐系统

笔记

MongoDB

文档数据库

Scala

Scala是一门多范式的编程语言，一种类似java的编程语言，设计初衷是实现可伸缩的语言、并集成面向对象编程和函数式编程的各种特性。

UGC

UGC 互联网术语，全称为User Generated Content，也就是用户生成内容，即用户原创内容。UGC的概念最早起源于互联网领域，即用户将自己原创的内容通过互联网平台进行展示或者提供给其他用户。UGC是伴随着以提倡个性化为主要特点的Web2.0概念而兴起的，也可叫做UCC（User-created Content）。它并不是某一种具体的业务，而是一种用户使用互联网的新方式，即由原来的以下载为主变成下载和上传并重。

随着互联网运用的发展，网络用户的交互作用得以体现，用户既是网络内容的浏览者，也是网络内容的创造者。

UDF

UDF（User-Defined Functions）即是用户定义的hive函数。hive自带的函数并不能完全满足业务需求，这时就需要我们自定义函数了

Spark

Spark是一种快速、通用、可扩展的大数据分析引擎，2009年诞生于加州大学伯克利分校AMPLab，2010年开源，2013年6月成为Apache孵化项目，2014年2月成为Apache顶级项目。目前，Spark生态系统已经发展成为一个包含多个子项目的集合，其中包含SparkSQL、Spark Streaming、GraphX、MLlib等子项目，Spark是基于内存计算的大数据并行计算框架。Spark基于内存计算，提高了在大数据环境下数据处理的实时性，同时保证了高容错性和高可伸缩性，允许用户将Spark部署在大量廉价硬件之上，形成集群。Spark得到了众多大数据公司的支持，这些公司包括Hortonworks、IBM、Intel、Cloudera、MapR、Pivotal、百度、阿里、腾讯、京东、携程、优酷土豆。当前百度的Spark已应用于凤巢、大搜索、直达号、百度大数据等业务；阿里利用GraphX构建了大规模的图计算和图挖掘系统，实现了很多生产系统的推荐算法；腾讯Spark集群达到8000台的规模，是当前已知的世界上最大的Spark集群。

为什么要学习Spark？

Hadoop的MapReduce计算模型存在的问题：

学习过Hadoop的MapReduce的学员都知道，MapReduce的核心是Shuffle（洗牌）。在整个Shuffle的过程中，至少会产生6次的I/O。下图是我们在讲MapReduce的时候，画的Shuffle的过程。

中间结果输出：基于MapReduce的计算引擎通常会将中间结果输出到磁盘上，进行存储和容错。另外，当一些查询（如：Hive）翻译到MapReduce任务时，往往会产生多个Stage（阶段），而这些串联的Stage又依赖于底层文件系统（如HDFS）来存储每一个Stage的输出结果，而I/O的效率往往较低，从而影响了MapReduce的运行速度。

Spark的最大特点：基于内存

Spark是MapReduce的替代方案，而且兼容HDFS、Hive，可融入Hadoop的生态系统，以弥补MapReduce的不足。

Spark的特点：快、易用、通用、兼容性

快

与Hadoop的MapReduce相比，Spark基于内存的运算速度要快100倍以上，即使，Spark基于硬盘的运算也要快10倍。Spark实现了高效的DAG执行引擎，从而可以通过内存来高效处理数据流。

易用

Spark支持Java、Python和Scala的API，还支持超过80种高级算法，使用户可以快速构建不同的应用。而且Spark支持交互式的Python和Scala的shell，可以非常方便地在这些shell中使用Spark集群来验证解决问题的方法。

通用

Spark提供了统一的解决方案。Spark可以用于批处理、交互式查询（Spark SQL）、实时流处理（Spark Streaming）、机器学习（Spark MLlib）和图计算（GraphX）。这些不同类型的处理都可以在同一个应用中无缝使用。Spark统一的解决方案非常具有吸引力，毕竟任何公司都想用统一的平台去处理遇到的问题，减少开发和维护的人力成本和部署平台的物力成本。

另外Spark还可以很好的融入Hadoop的体系结构中可以直接操作HDFS，并提供Hive on Spark、Pig on Spark的框架集成Hadoop。

兼容性

Spark可以非常方便地与其他的开源产品进行融合。比如，Spark可以使用Hadoop的YARN和Apache Mesos作为它的资源管理和调度器，器，并且可以处理所有Hadoop支持的数据，包括HDFS、HBase和Cassandra等。这对于已经部署Hadoop集群的用户特别重要，因为不需要做任何数据迁移就可以使用Spark的强大处理能力。Spark也可以不依赖于第三方的资源管理和调度器，它实现了Standalone作为其内置的资源管理和调度框架，这样进一步降低了Spark的使用门槛，使得所有人都可以非常容易地部署和使用Spark。此外，Spark还提供了在EC2上部署Standalone的Spark集群的工具。

Spark Core

主要功能

Spark Core提供Spark最基础与最核心的功能，主要包括以下功能：

(1) SparkContext：通常而言，Driver Application的执行与输出都是通过SparkContext来完成的。在正式提交Application之前，首先需要初始化SparkContext。SparkContext隐藏了网络通信、分布式部署、消息通信、存储能力、计算能力、缓存、测量系统、文件服务、Web服务等内容，应用程序开发者只需要使用SparkContext提供的API完成功能开发。SparkContext内置的DAGScheduler负责创建Job，将DAG中的RDD划分到不同的Stage，提交Stage等功能。内置的TaskScheduler负责资源的申请，任务的提交及请求集群对任务的调度等工作。
(2) 存储体系：Spark优先考虑使用各节点的内存作为存储，当内存不足时才会考虑使用磁盘，这极大地减少了磁盘IO，提升了任务执行的效率，使得Spark适用于实时计算、流式计算等场景。此外，Spark还提供了以内存为中心的高容错的分布式文件系统Tachyon供用户进行选择。Tachyon能够为Spark提供可靠的内存级的文件共享服务。
(3) 计算引擎：计算引擎由SparkContext中的DAGScheduler、RDD以及具体节点上的Executor负责执行的Map和Reduce任务组成。DAGScheduler和RDD虽然位于SparkContext内部，但是在任务正式提交与执行之前会将Job中的RDD组织成有向无环图（DAG），并对Stage进行划分，决定了任务执行阶段任务的数量、迭代计算、shuffle等过程。
(4) 部署模式：由于单节点不足以提供足够的存储和计算能力，所以作为大数据处理的Spark在SparkContext的TaskScheduler组件中提供了对Standalone部署模式的实现和Yarn、Mesos等分布式资源管理系统的支持。通过使用Standalone、Yarn、Mesos等部署模式为Task分配计算资源，提高任务的并发执行效率。

Spark Core子框架

Spark的几大子框架包括：

(1)、Spark SQL：首先使用SQL语句解析器（SqlParser）将SQL转换为语法树（Tree），并且使用规则执行器（RuleExecutor）将一系列规则（Rule）应用到语法树，最终生成物理执行计划并执行。其中，规则执行器包括语法分析器（Analyzer）和优化器（Optimizer）。
(2)、Spark Streaming：用于流式计算。Spark Streaming支持Kafka、Flume、Twitter、MQTT、ZeroMQ、Kinesis和简单的TCP套接字等多种数据输入源。输入流接收器（Receiver）负责接入数据，是接入数据流的接口规范。Dstream是Spark Streaming中所有数据流的抽象，Dstream可以被组织为Dstream Graph。Dstream本质上由一系列连续的RDD组成。
(3)、GraphX：Spark提供的分布式图计算框架。GraphX主要遵循整体同步并行（bulk Synchronous parallel，BSP）计算模式下的Pregel模型实现。GraphX提供了对图的抽象Graph，Graph由顶点（Vertex），边（Edge）及继承了Edge的EdgeTriplet三种结构组成。GraphX目前已经封装了最短路径，网页排名，连接组件，三角关系统计等算法的实现，用户可以选择使用。
(4)、MLlib：Spark提供的机器学习框架。机器学习是一门设计概率论、统计学、逼近论、凸分析、算法复杂度理论等多领域的交叉学科。MLlib目前已经提供了基础统计、分析、回归、决策树、随机森林、朴素贝叶斯、保序回归、协同过滤、聚类、维数缩减、特征提取与转型、频繁模式挖掘、预言模型标记语言、管道等多种数理统计、概率论、数据挖掘方面的数学算法。

Spark架构

Spark采用了分布式计算中的Master-Slave模型。Master作为整个集群的控制器，负责整个集群的正常运行；Worker是计算节点，接受主节点命令以及进行状态汇报；Executor负责任务（Tast）的调度和执行；Client作为用户的客户端负责提交应用；Driver负责控制一个应用的执行。

Spark集群启动时，需要从主节点和从节点分别启动Master进程和Worker进程，对整个集群进行控制。在一个Spark应用的执行过程中，Driver是应用的逻辑执行起点，运行Application的main函数并创建SparkContext，DAGScheduler把对Job中的RDD有向无环图根据依赖关系划分为多个Stage，每一个Stage是一个TaskSet， TaskScheduler把Task分发给Worker中的Executor；Worker启动Executor，Executor启动线程池用于执行Task。

Spark计算模型

RDD：弹性分布式数据集，是一种内存抽象，可以理解为一个大数组，数组的元素是RDD的分区Partition，分布在集群上；在物理数据存储上，RDD的每一个Partition对应的就是一个数据块Block，Block可以存储在内存中，当内存不够时可以存储在磁盘上。

RDD逻辑物理结构

Hadoop将Mapreduce计算的结果写入磁盘，在机器学习、图计算、PageRank等迭代计算下，重用中间结果导致的反复I/O耗时过长，成为了计算性能的瓶颈。为了提高迭代计算的性能和分布式并行计算下共享数据的容错性，伯克利的设计者依据两个特性而设计了RDD:

1、数据集分区存储在节点的内存中，减少迭代过程（如机器学习算法）反复的I/O操作从而提高性能。
2、数据集不可变，并记录其转换过程，从而实现无共享数据读写同步问题、以及出错的可重算性。

Operations：算子

算子是RDD中定义的函数，可以对RDD中的数据进行转换和操作。如下图，Spark从外部空间（HDFS）读取数据形成RDD_0，Tranformation算子对数据进行操作（如fliter）并转化为新的RDD_1、RDD_2，通过Action算子（如collect/count）触发Spark提交作业。

如上的分析过程可以看出，Tranformation算子并不会触发Spark提交作业，直至Action算子才提交作业，这是一个延迟计算的设计技巧，可以避免内存过快被中间计算占满，从而提高内存的利用率。

下图是算子的列表，分三大类：Value数据类型的Tranformation算子；Key-Value数据类型的Tranformation算子；Action算子。

Lineage Graph：血统关系图

下图的第一阶段生成RDD的有向无环图，即是血统关系图，记录了RDD的更新过程，当这个RDD的部分分区数据丢失时，它可以通过Lineage获取足够的信息来重新运算和恢复丢失的数据分区。DAGScheduler依据RDD的依赖关系将有向无环图划分为多个Stage，一个Stage对应着一系列的Task，由TashScheduler分发给Worker计算。

组件

介绍

spark生态系统中，Spark Core，包括各种Spark的各种核心组件，它们能够对内存和硬盘进行操作，或者调用CPU进行计算。
spark core定义了RDD、DataFrame和DataSet

spark最初只有RDD，DataFrame在Spark 1.3中被首次发布，DataSet在Spark1.6版本中被加入。

RDD

RDD：Spark的核心概念是RDD (resilient distributed dataset)，指的是一个只读的，可分区的分布式数据集，这个数据集的全部或部分可以缓存在内存中，在多次计算间重用。

import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}
 
object Run {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("test").setMaster("local")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")
    val sqlContext = new SQLContext(sc)
 
    /**
      * id      age
      * 1       30
      * 2       29
      * 3       21
      */
    case class Person(id: Int, age: Int)
    val idAgeRDDPerson = sc.parallelize(Array(Person(1, 30), Person(2, 29), Person(3, 21)))
 
    // 优点1
    // idAge.filter(_.age > "") // 编译时报错, int不能跟String比
 
    // 优点2
    idAgeRDDPerson.filter(_.age > 25) // 直接操作一个个的person对象
  }
}
1234567891011121314151617181920212223242526

DataFrame

在Spark中，DataFrame是一种以RDD为基础的分布式数据集，类似于传统数据库中的二维表格。DataFrame与RDD的主要区别在于，前者带有schema元信息，即DataFrame所表示的二维表数据集的每一列都带有名称和类型。这使得Spark SQL得以洞察更多的结构信息，从而对藏于DataFrame背后的数据源以及作用于DataFrame之上的变换进行了针对性的优化，最终达到大幅提升运行时效率的目标。反观RDD，由于无从得知所存数据元素的具体内部结构，Spark Core只能在stage层面进行简单、通用的流水线优化。

DataFrame引入了schema和off-heap

schema : RDD每一行的数据, 结构都是一样的.
这个结构就存储在schema中。 Spark通过schame就能够读懂数据， 因此在通信和IO时就只需要序列化和反序列化数据，而结构的部分就可以省略了。 off-heap : 意味着JVM堆以外的内存，这些内存直接受操作系统管理（而不是JVM）。Spark能够以二进制的形式序列化数据(不包括结构)到off-heap中，当要操作数据时，就直接操作off-heap内存。由于Spark理解schema，所以知道该如何操作。

off-heap就像地盘，schema就像地图， Spark有地图又有自己地盘了， 就可以自己说了算了， 不再受JVM的限制，也就不再收GC的困扰了。通过schema和off-heap，DataFrame解决了RDD的缺点，但是却丢了RDD的优点。 DataFrame不是类型安全的， API也不是面向对象风格的。

import org.apache.spark.sql.types.{DataTypes, StructField, StructType}
import org.apache.spark.sql.{Row, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}
 
object Run {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("test").setMaster("local")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")
    val sqlContext = new SQLContext(sc)
    /**
      * id      age
      * 1       30
      * 2       29
      * 3       21
      */
    val idAgeRDDRow = sc.parallelize(Array(Row(1, 30), Row(2, 29), Row(4, 21)))
 
    val schema = StructType(Array(StructField("id", DataTypes.IntegerType), StructField("age", DataTypes.IntegerType)))
 
    val idAgeDF = sqlContext.createDataFrame(idAgeRDDRow, schema)
    // API不是面向对象的
    idAgeDF.filter(idAgeDF.col("age") > 25) 
    // 不会报错, DataFrame不是编译时类型安全的
    idAgeDF.filter(idAgeDF.col("age") > "") 
  }
}
123456789101112131415161718192021222324252627

DataSet

Dataset是一个强类型的特定领域的对象，这种对象可以函数式或者关系操作并行地转换。每个Dataset也有一个被称为一个DataFrame的类型化视图，这种DataFrame是Row类型的Dataset，即Dataset[Row]

Dataset是“懒惰”的，只在执行行动操作时触发计算。本质上，数据集表示一个逻辑计划，该计划描述了产生数据所需的计算。当执行行动操作时，Spark的查询优化程序优化逻辑计划，并生成一个高效的并行和分布式物理计划。

DataSet结合了RDD和DataFrame的优点,，并带来的一个新的概念Encoder 当序列化数据时，Encoder产生字节码与off-heap进行交互，能够达到按需访问数据的效果， 而不用反序列化整个对象。 Spark还没有提供自定义Encoder的API，但是未来会加入。

下面看DataFrame和DataSet在2.0.0-preview中的实现

下面这段代码, 在1.6.x中创建的是DataFrame
// 上文DataFrame示例中提取出来的
val idAgeRDDRow = sc.parallelize(Array(Row(1, 30), Row(2, 29), Row(4, 21)))
 
val schema = StructType(Array(StructField("id", DataTypes.IntegerType), StructField("age", DataTypes.IntegerType)))
 
val idAgeDF = sqlContext.createDataFrame(idAgeRDDRow, schema)
1234567
但是同样的代码在2.0.0-preview中, 创建的虽然还叫DataFrame

// sqlContext.createDataFrame(idAgeRDDRow, schema) 方法的实现, 返回值依然是DataFrame
def createDataFrame(rowRDD: RDD[Row], schema: StructType): DataFrame = {
sparkSession.createDataFrame(rowRDD, schema)
}
123456
但是其实却是DataSet, 因为DataFrame被声明为Dataset[Row]

package object sql {
  // ...省略了不相关的代码
 
  type DataFrame = Dataset[Row]
}
1234567
因此当我们从1.6.x迁移到2.0.0的时候, 无需任何修改就直接用上了DataSet.

下面是一段DataSet的示例代码

import org.apache.spark.sql.types.{DataTypes, StructField, StructType}
import org.apache.spark.sql.{Row, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}
 
object Test {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("test").setMaster("local") // 调试的时候一定不要用local[*]
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    import sqlContext.implicits._
 
    val idAgeRDDRow = sc.parallelize(Array(Row(1, 30), Row(2, 29), Row(4, 21)))
 
    val schema = StructType(Array(StructField("id", DataTypes.IntegerType), StructField("age", DataTypes.IntegerType)))
 
    // 在2.0.0-preview中这行代码创建出的DataFrame, 其实是DataSet[Row]
    val idAgeDS = sqlContext.createDataFrame(idAgeRDDRow, schema)
 
    // 在2.0.0-preview中, 还不支持自定的Encoder, Row类型不行, 自定义的bean也不行
    // 官方文档也有写通过bean创建Dataset的例子，但是我运行时并不能成功
    // 所以目前需要用创建DataFrame的方法, 来创建DataSet[Row]
    // sqlContext.createDataset(idAgeRDDRow)
 
    // 目前支持String, Integer, Long等类型直接创建Dataset
    Seq(1, 2, 3).toDS().show()
    sqlContext.createDataset(sc.parallelize(Array(1, 2, 3))).show()
  }
}
1234567891011121314151617181920212223242526272829303132

RDD和DataFrame比较

DataFrame与RDD相同之处，都是不可变分布式弹性数据集。不同之处在于，DataFrame的数据集都是按指定列存储，即结构化数据。类似于传统数据库中的表。
DataFrame的设计是为了让大数据处理起来更容易。DataFrame允许开发者把结构化数据集导入DataFrame，并做了higher-level的抽象; DataFrame提供特定领域的语言（DSL）API来操作你的数据集。

上图直观地体现了DataFrame和RDD的区别。左侧的RDD[Person]虽然以Person为类型参数，但Spark框架本身不了解Person类的内部结构。而右侧的DataFrame却提供了详细的结构信息，使得Spark SQL可以清楚地知道该数据集中包含哪些列，每列的名称和类型各是什么。DataFrame多了数据的结构信息，即schema。RDD是分布式的Java对象的集合。DataFrame是分布式的Row对象的集合。DataFrame除了提供了比RDD更丰富的算子以外，更重要的特点是提升执行效率、减少数据读取以及执行计划的优化，比如filter下推、裁剪等。

RDD和DataSet比较

DataSet以Catalyst逻辑执行计划表示，并且数据以编码的二进制形式被存储，不需要反序列化就可以执行sorting、shuffle等操作。

DataSet创立需要一个显式的Encoder，把对象序列化为二进制，可以把对象的scheme映射为Spark SQl类型，然而RDD依赖于运行时反射机制。

通过上面两点，DataSet的性能比RDD的要好很多

DataFrame和DataSet比较

Dataset可以认为是DataFrame的一个特例，主要区别是Dataset每一个record存储的是一个强类型值而不是一个Row。因此具有如下三个特点：

1.DataSet可以在编译时检查类型
2.是面向对象的编程接口。用wordcount举例：
3.后面版本DataFrame会继承DataSet，DataFrame是面向Spark SQL的接口。
DataFrame和DataSet可以相互转化， df.as[ElementType]这样可以把DataFrame转化为DataSet，ds.toDF()这样可以把DataSet转化为DataFrame。

//DataFrame
 
// Load a text file and interpret each line as a java.lang.String
val ds = sqlContext.read.text("/home/spark/1.6/lines").as[String]
val result = ds
  .flatMap(_.split(" "))               // Split on whitespace
  .filter(_ != "")                     // Filter empty words
  .toDF()                              // Convert to DataFrame to perform aggregation / sorting
  .groupBy($"value")                   // Count number of occurences of each word
  .agg(count("*") as "numOccurances")
  .orderBy($"numOccurances" desc)      // Show most common words first

//DataSet,完全使用scala编程，不要切换到DataFrame
 
val wordCount =
  ds.flatMap(_.split(" "))
    .filter(_ != "")
    .groupBy(_.toLowerCase()) // Instead of grouping on a column expression (i.e. $"value") we pass a lambda function
    .count()

1234567891011121314151617181920

应用场景

什么时候用RDD？使用RDD的一般场景:

你需要使用low-level的transformation和action来控制你的数据集；
你得数据集非结构化，比如，流媒体或者文本流；
你想使用函数式编程来操作你得数据，而不是用特定领域语言（DSL）表达；
你不在乎schema，比如，当通过名字或者列处理（或访问）数据属性不在意列式存储格式；
你放弃使用DataFrame和Dataset来优化结构化和半结构化数据集
RDD在Apache Spark 2.0中惨遭抛弃？
答案当然是 NO !
通过后面的描述你会得知：Spark用户可以在RDD，DataFrame和Dataset三种数据集之间无缝转换，而是只需使用超级简单的API方法。

什么时候使用DataFrame或者Dataset？

你想使用丰富的语义，high-level抽象，和特定领域语言API，那你可DataFrame或者Dataset；
你处理的半结构化数据集需要high-level表达， filter，map，aggregation，average，sum ，SQL 查询，列式访问和使用lambda函数，那你可DataFrame或者Dataset；
你想利用编译时高度的type-safety，Catalyst优化和Tungsten的code生成，那你可DataFrame或者Dataset；
你想统一和简化API使用跨Spark的Library，那你可DataFrame或者Dataset；
如果你是一个R使用者，那你可DataFrame或者Dataset；
如果你是一个Python使用者，那你可DataFrame或者Dataset；

你可以无缝的把DataFrame或者Dataset转化成一个RDD，只需简单的调用 .rdd：

// select specific fields from the Dataset, apply a predicate
// using the where() method, convert to an RDD, and show first 10
// RDD rows
 
val deviceEventsDS = ds.select($"device_name", $"cca3", $"c02_level").where($"c02_level" > 1300)
// convert to RDDs and take the first 10 rows
 
val eventsRDD = deviceEventsDS.rdd.take(10)

Spark SQL

​ spark SQL是spark的一个模块，主要用于进行结构化数据的处理。它提供的最核心的编程抽象就是DataFrame。

Spark Streaming

Spark Streaming是核心Spark API的扩展，可实现可扩展、高吞吐量、可容错的实时数据流处理。数据可以从诸如Kafka，Flume，Kinesis或TCP套接字等众多来源获取，并且可以使用由高级函数（如map，reduce，join和window）开发的复杂算法进行流数据处理。最后，处理后的数据可以被推送到文件系统，数据库和实时仪表板。而且，您还可以在数据流上应用Spark提供的机器学习和图处理算法。

Flume-ng

Flume是一个分布式、可靠、高可用的海量日志聚合系统，支持在系统中定制各类数据发送方，用于收集数据；同时，Flume提供对数据的简单处理，并写到各种数据接收方的能力。

Flume在0.9.x和1.x之间有较大的架构调整，1.x版本之后的改称为Flume NG。0.9.x的称为Flume OG。

Flume OG体系架构如下，Flume OG已经不再进行版本更新：

Kafka

概述

Kafka是最初由Linkedin公司开发，是一个分布式、分区的、多副本的、多订阅者，基于zookeeper协调的分布式日志系统（也可以当做MQ系统），常见可以用于web/nginx日志、访问日志，消息服务等等，Linkedin于2010年贡献给了Apache基金会并成为顶级开源项目。

主要应用场景是：日志收集系统和消息系统。

Kafka主要设计目标如下：

• 以时间复杂度为O(1)的方式提供消息持久化能力，即使对TB级以上数据也能保证常数时间的访问性能。
• 高吞吐率。即使在非常廉价的商用机器上也能做到单机支持每秒100K条消息的传输。
• 支持Kafka Server间的消息分区，及分布式消费，同时保证每个partition内的消息顺序传输。
• 同时支持离线数据处理和实时数据处理。
• Scale out:支持在线水平扩展

核心概念

名词	解释
Producer	消息的生成者
Consumer	消息的消费者
ConsumerGroup	消费者组，可以并行消费Topic中的partition的消息
Broker	缓存代理，Kafka集群中的一台或多台服务器统称broker.
Topic	Kafka处理资源的消息源(feeds of messages)的不同分类
Partition	Topic物理上的分组，一个topic可以分为多个partion,每个partion是一个有序的队列。partion中每条消息都会被分配一个 有序的Id(offset)
Message	消息，是通信的基本单位，每个producer可以向一个topic（主题）发布一些消息
Producers	消息和数据生成者，向Kafka的一个topic发布消息的 过程叫做producers
Consumers	消息和数据的消费者，订阅topic并处理其发布的消费过程叫做consumers

• 3.1 Producers的概念

1. 消息和数据生成者，向Kafka的一个topic发布消息的过程叫做producers
2. Producer将消息发布到指定的Topic中，同时Producer也能决定将此消息归属于哪个partition；比如基于round-robin方式 或者通过其他的一些算法等；
3. 异步发送批量发送可以很有效的提高发送效率。kafka producer的异步发送模式允许进行批量发送，先将消息缓存到内存中，然后一次请求批量发送出去。

• 3.2 broker的概念:

1. Broker没有副本机制，一旦broker宕机，该broker的消息将都不可用。
2. Broker不保存订阅者的状态，由订阅者自己保存。
3. 无状态导致消息的删除成为难题（可能删除的消息正在被订阅），Kafka采用基于时间的SLA（服务保证），消息保存一定时间（通常7天）后会删除。
4. 消费订阅者可以rewind back到任意位置重新进行消费，当订阅者故障时，可以选择最小的offset(id)进行重新读取消费消息

• 3.3 Message组成

1. Message消息：是通信的基本单位，每个producer可以向一个topic发布消息。
2. Kafka中的Message是以topic为基本单位组织的，不同的topic之间是相互独立的，每个topic又可以分成不同的partition每个partition储存一部分
3. partion中的每条Message包含以下三个属性：

offset	long
MessageSize	int32
data	messages的具体内容

• 3.4 Consumers的概念

  消息和数据消费者，订阅topic并处理其发布的消息的过程叫做consumers.
  在kafka中，我们可以认为一个group是一个“订阅者”，一个topic中的每个partions只会被一个“订阅者”中的一个consumer
  消费，不过一个consumer可以消费多个partitions中的消息
  注:
  Kafka的设计原理决定，对于一个topic，同一个group不能多于partition个数的consumer同时消费，否则将意味着某些 consumer无法得到消息

• 3.5 关键术语

主题，分区和偏移

主题是特定的数据流。它与NoSQL数据库中的表非常相似。与NoSQL数据库中的表一样，该主题被拆分为分区，使主题能够分布在各个节点上。与表中的主键一样，主题具有每个分区的偏移量。您可以使用其主题，分区和偏移量唯一标识消息。

分区

分区使主题可以在群集中分布。分区是水平可伸缩性的并行度单位。一个主题可以跨节点进行多个分区扩展。

消息根据分区键分配给分区; 如果没有分区键，则随机分配该分区。使用正确的密钥来避免热点非常重要。

分区中的每个消息都被分配一个称为偏移量的增量ID。每个分区的偏移量是唯一的，消息只在分区内排序。写入分区的消息是不可变的。

消息系统介绍

一个消息系统负责将数据从一个应用传递到另外一个应用，应用只需关注于数据，无需关注数据在两个或多个应用间是如何传递的。分布式消息传递基于可靠的消息队列，在客户端应用和消息系统之间异步传递消息。有两种主要的消息传递模式：点对点传递模式、发布-订阅模式。大部分的消息系统选用发布-订阅模式。Kafka就是一种发布-订阅模式。

点对点消息传递模式

在点对点消息系统中，消息持久化到一个队列中。此时，将有一个或多个消费者消费队列中的数据。但是一条消息只能被消费一次。当一个消费者消费了队列中的某条数据之后，该条数据则从消息队列中删除。该模式即使有多个消费者同时消费数据，也能保证数据处理的顺序。这种架构描述示意图如下：

生产者发送一条消息到queue，只有一个消费者能收到。

发布-订阅消息传递模式

在发布-订阅消息系统中，消息被持久化到一个topic中。与点对点消息系统不同的是，消费者可以订阅一个或多个topic，消费者可以消费该topic中所有的数据，同一条数据可以被多个消费者消费，数据被消费后不会立马删除。在发布-订阅消息系统中，消息的生产者称为发布者，消费者称为订阅者。该模式的示例图如下：

发布者发送到topic的消息，只有订阅了topic的订阅者才会收到消息。

Kafka的优点

解耦

在项目启动之初来预测将来项目会碰到什么需求，是极其困难的。消息系统在处理过程中间插入了一个隐含的、基于数据的接口层，两边的处理过程都要实现这一接口。这允许你独立的扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束。

冗余（副本）

有些情况下，处理数据的过程会失败。除非数据被持久化，否则将造成丢失。消息队列把数据进行持久化直到它们已经被完全处理，通过这一方式规避了数据丢失风险。许多消息队列所采用的”插入-获取-删除”范式中，在把一个消息从队列中删除之前，需要你的处理系统明确的指出该消息已经被处理完毕，从而确保你的数据被安全的保存直到你使用完毕。

扩展性

因为消息队列解耦了你的处理过程，所以增大消息入队和处理的频率是很容易的，只要另外增加处理过程即可。不需要改变代码、不需要调节参数。扩展就像调大电力按钮一样简单。

灵活性&峰值处理能力

在访问量剧增的情况下，应用仍然需要继续发挥作用，但是这样的突发流量并不常见；如果为以能处理这类峰值访问为标准来投入资源随时待命无疑是巨大的浪费。使用消息队列能够使关键组件顶住突发的访问压力，而不会因为突发的超负荷的请求而完全崩溃。

可恢复性

系统的一部分组件失效时，不会影响到整个系统。消息队列降低了进程间的耦合度，所以即使一个处理消息的进程挂掉，加入队列中的消息仍然可以在系统恢复后被处理。

顺序保证

在大多使用场景下，数据处理的顺序都很重要。大部分消息队列本来就是排序的，并且能保证数据会按照特定的顺序来处理。Kafka保证一个Partition内的消息的有序性。

缓冲

在任何重要的系统中，都会有需要不同的处理时间的元素。例如，加载一张图片比应用过滤器花费更少的时间。消息队列通过一个缓冲层来帮助任务最高效率的执行———写入队列的处理会尽可能的快速。该缓冲有助于控制和优化数据流经过系统的速度。

异步通信

很多时候，用户不想也不需要立即处理消息。消息队列提供了异步处理机制，允许用户把一个消息放入队列，但并不立即处理它。想向队列中放入多少消息就放多少，然后在需要的时候再去处理它们。

常用Message Queue对比

RabbitMQ

RabbitMQ是使用Erlang编写的一个开源的消息队列，本身支持很多的协议：AMQP，XMPP, SMTP, STOMP，也正因如此，它非常重量级，更适合于企业级的开发。同时实现了Broker构架，这意味着消息在发送给客户端时先在中心队列排队。对路由，负载均衡或者数据持久化都有很好的支持。

Redis

Redis是一个基于Key-Value对的NoSQL数据库，开发维护很活跃。虽然它是一个Key-Value数据库存储系统，但它本身支持MQ功能，所以完全可以当做一个轻量级的队列服务来使用。对于RabbitMQ和Redis的入队和出队操作，各执行100万次，每10万次记录一次执行时间。测试数据分为128Bytes、512Bytes、1K和10K四个不同大小的数据。实验表明：入队时，当数据比较小时Redis的性能要高于RabbitMQ，而如果数据大小超过了10K，Redis则慢的无法忍受；出队时，无论数据大小，Redis都表现出非常好的性能，而RabbitMQ的出队性能则远低于Redis。

ZeroMQ

ZeroMQ号称最快的消息队列系统，尤其针对大吞吐量的需求场景。ZeroMQ能够实现RabbitMQ不擅长的高级/复杂的队列，但是开发人员需要自己组合多种技术框架，技术上的复杂度是对这MQ能够应用成功的挑战。ZeroMQ具有一个独特的非中间件的模式，你不需要安装和运行一个消息服务器或中间件，因为你的应用程序将扮演这个服务器角色。你只需要简单的引用ZeroMQ程序库，可以使用NuGet安装，然后你就可以愉快的在应用程序之间发送消息了。但是ZeroMQ仅提供非持久性的队列，也就是说如果宕机，数据将会丢失。其中，Twitter的Storm 0.9.0以前的版本中默认使用ZeroMQ作为数据流的传输（Storm从0.9版本开始同时支持ZeroMQ和Netty作为传输模块）。

ActiveMQ

ActiveMQ是Apache下的一个子项目。 类似于ZeroMQ，它能够以代理人和点对点的技术实现队列。同时类似于RabbitMQ，它少量代码就可以高效地实现高级应用场景。

Kafka/Jafka

Kafka是Apache下的一个子项目，是一个高性能跨语言分布式发布/订阅消息队列系统，而Jafka是在Kafka之上孵化而来的，即Kafka的一个升级版。具有以下特性：快速持久化，可以在O(1)的系统开销下进行消息持久化；高吞吐，在一台普通的服务器上既可以达到10W/s的吞吐速率；完全的分布式系统，Broker、Producer、Consumer都原生自动支持分布式，自动实现负载均衡；支持Hadoop数据并行加载，对于像Hadoop的一样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。Kafka通过Hadoop的并行加载机制统一了在线和离线的消息处理。Apache Kafka相对于ActiveMQ是一个非常轻量级的消息系统，除了性能非常好之外，还是一个工作良好的分布式系统。

算法笔记

统计推荐

主要就是通过spark.sql使用查询语句进行排行的

历史热门商品推荐

根据评价次数进行排行

近期热门

按月进行热门热门商品推荐

平均评分成绩排行

对通过商品的评分取均值进行排行

离线推荐

ALS算法

隐语义模型是一种比较常用的协同过滤算法，基本思想是对稀疏矩阵进行模型分解，评估出缺失项的值，以此来得到一个基本的训练模型。然后依照此模型可以针对新的用户和物品数据进行评估。ALS是采用交替的最小二乘法来算出缺失项的。交替的最小二乘法是在最小二乘法的基础上发展而来的。

协同过滤算法，用户特征，商品特征，推荐

模型评估和参数选取

均方根误差

特征向量

tpi 商品p的特征向量， tqi商品q的特征向量

基于商品的特征向量，计算商品的相似度，进行推荐

实时推荐

背景：用户对产品的偏好随时间的推移总是会改变

要求：

1. 用户评分后或者最近几次评分后系统可以更新推荐结果

2. 计算快，准确率可适当降低

推荐算法公式：

Rr表示用户u 对商品r 的评分；

sim(q,r)表示商品q 与商品r 的相似度，设定最小相似度为0.6，当商品q和商品r 相似度低于0.6 的阈值，则视为两者不相关并忽略；

sim_sum 表示q 与RK 中商品相似度大于最小阈值的个数；

incount 表示RK 中与商品q 相似的、且本身评分较高（>=3）的商品个数；

recount 表示RK 中与商品q 相似的、且本身评分较低（<3）的商品个数；

公式前部分：对于每个候选商品q，从u 最近的K 个评分中，找出与q 相似度较高（>=0.6）的u 已评分商品们（复用离线的相似表），对于这些商品们中的每个商品r，将r 与q 的相似度乘以用户u 对r 的评分，将这些乘积计算平均数，作为用户u 对商品q 的评分预测

后半部分（奖惩因子）：

incount：跟候选商品q相似且评分大于某一阈值（>=3)的个数

与候选商品q的相似度越高还评分越高的，我们应该更大力度的推荐，优先级更高。