大数据-SparkStreaming（九）

2024-02-28 07:59

文章标签 数据 sparkstreaming

本文主要是介绍大数据-SparkStreaming（九），希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

大数据-SparkStreaming（九）

SparkStreaming调优

调整BlockReceiver的数量

案例演示：

val kafkaStream = {  val sparkStreamingConsumerGroup = "spark-streaming-consumer-group"  val kafkaParams = Map(  "zookeeper.connect" -> "node01:2181,node02:2181,node03:2181",  "group.id" -> "spark-streaming-test")  val inputTopic = "test"  val numPartitionsOfInputTopic = 3  val streams = (1 to numPartitionsOfInputTopic) map  {x =>  KafkaUtils.createStream(ssc, kafkaParams, Map(inputTopic -> 1),      StorageLevel.MEMORY_ONLY_SER).map(_._2)  }  val unifiedStream = ssc.union(streams)

调整Block的数量

batchInterval : 触发批处理的时间间隔
blockInterval :将接收到的数据生成Block的时间间隔，spark.streaming.blockInterval(默认是200ms)，那么，BlockRDD的分区数 = batchInterval / blockInterval，即一个Block就是RDD的一个分区，就是一个task
比如，batchInterval是2秒，而blockInterval是200ms，那么task数为10，如果task的数量太少，比一个executor的core数还少的话，那么可以减少blockInterval，blockInterval最好不要小于50ms，太小的话导致task数太多，那么launch task的时间久多了。

调整Receiver的接受速率

pps:permits per second 每秒允许接受的数据量(QPS -> queries per second)
Spark Streaming默认的PPS是没有限制的,可以通过参数spark.streaming.receiver.maxRate来控制，默认是Long.Maxvalue

调整数据处理的并行度

BlockRDD的分区数

a. 通过Receiver接受数据的特点决定

b. 也可以自己通过repartition设置

ShuffleRDD的分区数

a. 默认的分区数为spark.default.parallelism(core的大小)

b. 通过我们自己设置决定

val wordCounts = words.map(x => (x, 1)).reduceByKey((a: Int, b: Int) => a + b, new HashPartitioner(10))

数据的序列化

SparkStreaming两种需要序列化的数据： a. 输入的数据：默认是以StorageLevel.MEMORY_AND_DISK_SER_2的形式存储在executor上的内存中 b. 缓存的数据：默认是以StorageLevel.MEMORY_ONLY_SER的形式存储的内存中使用Kryo序列化机制，比Java序列化机制性能好

val conf = new SparkConf().setMaster(...).setAppName(...)
conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
conf.registerKryoClasses(Array(classOf[MyClass1], classOf[MyClass2]))
val sc = new SparkContext(conf)