大数据技术周报第 003 期

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

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
  }
}

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

// 设置任务名称
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)

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

 /**
  * 实例化一个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))

/**
 * 读取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(","))