这里记录过去一周，大数据相关值得分享的东西，每周日发布。

今天尝试写第三期，记录过去一周一点所见所闻。上周好像忘记发了？是的…

文章目录

技术一瞥
图片
文章
资源
视频
订阅

技术一瞥

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，如下图所示:

def partition(key: Any, numPartitions: Int): Int = {Utils.abs(key.hashCode) % numPartitions
}

这就保证了相同key的消息一定会被路由到相同的分区。如果你没有指定key，那么Kafka是如何确定这条消息去往哪个分区的呢？

if(key == null) {  // 如果没有指定keyval id = sendPartitionPerTopicCache.get(topic)  // 先看看Kafka有没有缓存的现成的分区Idid match {case Some(partitionId) =>  partitionId  // 如果有的话直接使用这个分区Id就好了case None => // 如果没有的话，val availablePartitions = topicPartitionList.filter(_.leaderBrokerIdOpt.isDefined)  //找出所有可用分区的leader所在的brokerif (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).partitionIdsendPartitionPerTopicCache.put(topic, partitionId) // 更新缓存以备下一次直接使用partitionId}
}

2、开发 Spark 流式程序，少不了写 scala 代码，其语法相当简介，如果不写注释一定会挨骂的！

官网Scala 文档
官网Spark 文档

编写一个流式的工具类：

Spark 相关：

protected val sparkConf = new SparkConf
protected var ssc: StreamingContext = _
protected var sc: SparkContext = _
protected var fs: FileSystem = _
protected var spark: SparkSession = _

设置 Spark 的一系列参数：

// 设置任务名称
def setJobName(name: String): SparkConf = {sparkConf.setAppName(name)
}
// 设置任务参数
def set(key: String, value: String): SparkConf = {sparkConf.set(key, value)
}

初始化任务：

// 初始化
spark = SparkSession.builder().config(sparkConf).getOrCreate()
sc = spark.sparkContext
ssc = new StreamingContext(spark.sparkContext, Seconds(sec))fs = FileSystem.get(ssc.sparkContext.hadoopConfiguration)

Zookeeper 相关：

// 获取配置文件的 zk 服务地址信息
protected val zkHosts: String = PropertiesUtils.get("zookeeper.hosts")
// zookeeper 客户端
protected lazy val zkClient = new ZkClient(zkHosts)

实例化 stream:

 /*** 实例化一个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 的一个目录

/*** 读取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).toMapcase 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、原则

2、房贷

原话是：
第一，30年等额本息，就是最优解。第二，能贷多久贷多久，不能提前还。第三，房贷不可怕，5年以后云淡风清。

3、周末

真相是：一边是 LOL, 一边是拖更的文章。

文章

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

视频

1、【z说球鞋】看懂球鞋经济学
本期视频35分钟，球鞋经济学内容，不到一节课时间，你将收获课堂上从没讲过的重要知识点。点击量可能很惨，但这不重要，说给有心人，越早听到，对你的人生也许越有帮助。

大数据技术周报第 009 期
这里记录过去一周,大数据相关值得分享的东西,每周日发布. 这是第 9 期分享,记录过去一周关于大数据的见闻,保证信息的质量和溯源. 文章目录技术一瞥文章资源订阅技术一瞥 1.Kafka 最近 ...
大数据技术周报第 007 期
这里记录过去一周,大数据相关值得分享的东西,每周发布. 今天尝试写第 7 期,记录过去一周一点所见所闻. 文章目录技术一瞥图片文章资源订阅本期主题:实时数仓技术一瞥做实时数仓经历了什么 ...
大数据技术周报第 006 期
这里记录过去一周,大数据相关值得分享的东西,每周发布. 今天尝试写第 6 期,记录过去一周一点所见所闻.假期玩(｡･∀･)ﾉﾞ嗨了. 文章目录技术一瞥图片文章资源订阅技术一瞥 1.hdfs ...
大数据技术周报第 005 期
这里记录过去一周,大数据相关值得分享的东西,每周发布. 今天尝试写第四期,记录过去一周一点所见所闻. 文章目录技术一瞥文章资源订阅技术一瞥 1.什么是ETL? 引用:https://en.w ...
大数据技术周报第 004 期
这里记录过去一周,大数据相关值得分享的东西,每周日发布. 今天尝试写第 4 期,记录过去一周一点所见所闻. 文章目录图片文章资源订阅图片文章 1.Flink 1.9 发布,有你期待的功能吗 ...
大数据技术周报第 002 期
这里记录过去一周,大数据相关值得分享的东西,每周日发布. 今天尝试写第二期,记录过去一周一点所见所闻.本周划水太严重了,主要还是陪女朋友比较重要=-= 文章目录图片文章资源摘录订阅图片来自 ...
大数据技术周报第 001 期
这里记录过去一周,大数据相关值得分享的东西,每周日发布. 今天尝试写第一期,记录过去一周一点所见所闻. 文章目录最佳技术一瞥图片文章资源订阅最佳 1.TDengine TDengine是 ...
【2017年第4期】工业大数据技术与架构
郑树泉,1,3, 覃海焕,2,3, 王倩,1,3 1. 上海计算机软件技术开发中心,上海 201112 2. 上海电机学院电子信息学院,上海 201306 3. 上海产业技术研究院工程大数据服务创新中 ...
《大数据》2015年第2期“前沿”——大数据技术发展的十个前沿方向（上）
大数据技术发展的十个前沿方向(上) 吴甘沙英特尔中国研究院 doi:10.11959/j.issn.2096-0271.2015023 Ten Fronties for Big Data Techn ...

大数据技术周报第 003 期

文章目录

技术一瞥

图片

文章

资源

视频

订阅

大数据技术周报第 003 期相关推荐

最新文章

热门文章