这里记录过去一周，大数据相关值得分享的东西，每周日发布。
今天尝试写第三期，记录过去一周一点所见所闻。上周好像忘记发了？是的...
技术一瞥
1、Kafka 的分区数？数据发送为什么要带上 key?
在回答第一个问题之前，先来看一段 kafka 的简单描述：
① Kafka官网上标榜自己是"high-throughput distributed messaging system"，即一个高吞吐量的分布式消息引擎。那么怎么达到高吞吐量呢？Kafka在底层摒弃了Java堆缓存机制，采用了操作系统级别的页缓存，同时将随机写操作改为顺序写，再结合Zero-Copy的特性极大地改善了IO性能。但是，这只是一个方面，毕竟单机优化的能力是有上限的。如何通过水平扩展甚至是线性扩展来进一步提升吞吐量呢？Kafka就是使用了分区(partition)，通过将topic的消息打散到多个分区并分布保存在不同的broker上实现了消息处理(不管是producer还是consumer)的高吞吐量。
② Kafka的生产者和消费者都可以多线程地并行操作，而每个线程处理的是一个分区的数据。因此**分区实际上是调优Kafka并行度的最小单元。**对于producer而言，它实际上是用多个线程并发地向不同分区所在的broker发起Socket连接同时给这些分区发送消息；而consumer呢，同一个消费组内的所有consumer线程都被指定topic的某一个分区进行消费(具体如何确定consumer线程数目我们后面会详细说明)。所以说，如果一个topic分区越多，理论上整个集群所能达到的吞吐量就越大。
但分区是否越多越好呢？显然也不是，因为每个分区都有自己的开销：
一是客户端、服务端需要的内存会变多（需要维护一些分区的信息，如果分区越多，这些信息所占的内存就越大）
二是文件句柄的开销（每个分区在底层文件系统都有属于自己的一个目录。该目录下通常会有两个文件：base_offset.log和base_offset.index。Kafak的controller和ReplicaManager会为每个broker都保存这两个文件句柄(file handler)。很明显，如果分区数越多，所需要保持打开状态的文件句柄数也就越多，最终可能会突破你的ulimit -n的限制。）
三是**降低了高可用 **（如果你有10000个分区，10个broker，也就是说平均每个broker上有1000个分区。此时这个broker挂掉了，那么zookeeper和controller需要立即对这1000个分区进行leader选举。比起很少的分区leader选举而言，这必然要花更长的时间，并且通常不是线性累加的。如果这个broker还同时是controller情况就更糟了）。
默认情况下，Kafka根据传递消息的key来进行分区的分配，即hash(key) % numPartitions，如下图所示:
代码语言：javascript复制def partition(key: Any, numPartitions: Int): Int = {
Utils.abs(key.hashCode) % numPartitions
}
这就保证了相同key的消息一定会被路由到相同的分区。如果你没有指定key，那么Kafka是如何确定这条消息去往哪个分区的呢？
代码语言：javascript复制if(key == null) { // 如果没有指定key
val id = sendPartitionPerTopicCache.get(topic) // 先看看Kafka有没有缓存的现成的分区Id
id match {
case Some(partitionId) =>
partitionId // 如果有的话直接使用这个分区Id就好了
case None => // 如果没有的话，
val availablePartitions = topicPartitionList.filter(_.leaderBrokerIdOpt.isDefined) //找出所有可用分区的leader所在的broker
if (availablePartitions.isEmpty)
throw new LeaderNotAvailableException("No leader for any partition in topic " topic)
val index = Utils.abs(Random.nextInt) % availablePartitions.size // 从中随机挑一个
val partitionId = availablePartitions(index).partitionId
sendPartitionPerTopicCache.put(topic, partitionId) // 更新缓存以备下一次直接使用
partitionId
}
}
2、开发 Spark 流式程序，少不了写 scala 代码，其语法相当简介，如果不写注释一定会挨骂的！
官网Scala 文档官网Spark 文档
编写一个流式的工具类：
Spark 相关：
代码语言：javascript复制protected val sparkConf = new SparkConf
protected var ssc: StreamingContext = _
protected var sc: SparkContext = _
protected var fs: FileSystem = _
protected var spark: SparkSession = _
设置 Spark 的一系列参数：
代码语言：javascript复制// 设置任务名称
def setJobName(name: String): SparkConf = {
sparkConf.setAppName(name)
}
// 设置任务参数
def set(key: String, value: String): SparkConf = {
sparkConf.set(key, value)
}
初始化任务：
代码语言：javascript复制// 初始化
spark = SparkSession.builder().config(sparkConf).getOrCreate()
sc = spark.sparkContext
ssc = new StreamingContext(spark.sparkContext, Seconds(sec))
fs = FileSystem.get(ssc.sparkContext.hadoopConfiguration)
Zookeeper 相关：
代码语言：javascript复制// 获取配置文件的 zk 服务地址信息
protected val zkHosts: String = PropertiesUtils.get("zookeeper.hosts")
// zookeeper 客户端
protected lazy val zkClient = new ZkClient(zkHosts)
实例化 stream:
代码语言：javascript复制 /**
* 实例化一个Stream
* @param kafkaParams 参数
* @param offsets 消费位移
* @param topicArr 主题数组
* @return
*/
protected def getStream(kafkaParams: Map[String, Object],offsets: Map[TopicPartition, Long],
topicArr: Array[String]): InputDStream[ConsumerRecord[String, String]] =
KafkaUtils.createDirectStream[String, String](ssc,
PreferConsistent,
// 1-首次启动；2-其他次启动
if (offsets.nonEmpty) Subscribe[String, String](topicArr, kafkaParams, offsets)
else Subscribe[String, String](topicArr, kafkaParams))
读取、保存 offset 到 zk 的一个目录
代码语言：javascript复制/**
* 读取zookeeper中保存的kafka主题消费位移，partition1:offset1,...,partitionN:offsetN
* Zookeeper保存Kafka数据消费位移，路径格式：/kafka/offsets/groupId/topic，数据格式：partitionId1:offset1,partitionId2:offset2,partitionId3:offset3
*
* @param zkClient zookeeper客户端
* @param zkPath zookeeper保存路径
* @param topic kafka主题
* @return
*/
protected def readOffsetsTopic(zkClient: ZkClient, zkPath: String, topic: String): Map[TopicPartition, Long] = {
val (offsetsRangesStrOpt, _) = ZkUtils.readDataMaybeNull(zkClient, zkPath s"/$topic")
offsetsRangesStrOpt match {
case Some(offsetsRangesStr) =>
offsetsRangesStr.split(',')
.map(s => s.split(':'))
.map(p => new TopicPartition(topic, p(0).toInt) -> p(1).toLong)
.toMap
case None => Map[TopicPartition, Long]()
}
}
/**
* 保存kafka主题消费位移到zookeeper中，partition1:offset1,...,partitionN:offsetN
* Zookeeper保存Kafka数据消费位移，路径格式：/kafka/offsets/groupId/topic，数据格式：partitionId1:offset1,partitionId2:offset2,partitionId3:offset3
*
* @param zkClient zookeeper客户端
* @param zkPath zookeeper保存路径
* @param topic kafka主题
* @param rdd 待保存位置的RDD
*/
protected def saveOffsetsTopic(zkClient: ZkClient, zkPath: String, topic: String, rdd: RDD[_]): Unit =
ZkUtils.updatePersistentPath(zkClient, zkPath s"/$topic", rdd.asInstanceOf[HasOffsetRanges].offsetRanges
.map(offsetRange => s"${offsetRange.partition}:${offsetRange.untilOffset}")
.mkString(","))
Scala 下划线主要用法：第一点，一个类型数据的默认值，譬如var i：Int = _，这里是0。整形为0，浮点为0.0，引用类型为null。第二点，匿名函数的参数，一个匿名函数里第一个下划线代表第一个参数，第二个代表第二个参数第三点，import的通配符第四点，重命名import时隐藏某个名称时的用法
第五点，模式匹配中代表会被丢弃的值
文章
1、Redis进阶实践之Redis和Lua初步整合使用lua这个脚本是一个好东西，可以运行在任何平台上，也可以嵌入到大多数语言当中，来扩展其功能。lua脚本是用C语言写的，体积很小，运行速度很快，并且每次的执行都是作为一个原子事务来执行的，我们可以在其中做很多的事情。
2、大家所推崇的Redis分布式锁真的就万无一失吗？在单实例JVM中，常见的处理并发问题的方法有很多，比如synchronized关键字进行访问控制、volatile关键字、ReentrantLock等常用方法。但是在分布式环境中，上述方法却不能在跨JVM场景中用于处理并发问题，当业务场景需要对分布式环境中的并发问题进行处理时，需要使用分布式锁来实现。
资源
1、GitHub大数据/数据挖掘/推荐系统/机器学习相关资源
以数据挖掘、机器学习为主。
2、官网lua 官方文档
它好的哪？
Redis在2.6推出了脚本功能，允许开发者使用Lua语言编写脚本传到Redis中执行。使用脚本的好处如下:
- 减少网络开销：本来5次网络请求的操作，可以用一个请求完成，原先5次请求的逻辑放在redis服务器上完成。使用脚本，减少了网络往返时延。
- 原子操作：Redis会将整个脚本作为一个整体执行，中间不会被其他命令插入。
- 复用：客户端发送的脚本会永久存储在Redis中，意味着其他客户端可以复用这一脚本而不需要使用代码完成同样的逻辑。
附上一个开源的 lua 脚本调试工具：开源ZeroBrane Studio is a lightweight Lua IDE
