SparkStreaming+Kafka+ES使用笔记

emm

非专业开发&＃xff0c;一点笔记

Kafka

1. 当Kafka内容特别大时会报错&＃xff0c;此时设置下fetch.message.max.bytes为一个比较大的值就好。

val kafkaParams &＃61; Map[String, String]("metadata.broker.list" -> brokers,"fetch.message.max.bytes" -> "10485760" )


2. 关于partitions
   KafkaDirectStream提供的分片数与Kafka的分片数相同
   如果运算量远远超过了重新分片的消耗&＃xff0c;可以用repartition&＃xff0c;否则还是增加同时运行的Jobs数量。

Spark Streaming

1. 增加同时运行的任务数量

SparkStreaming默认只启动一个Job&＃xff0c;所以使用核心再多如果任务数量不够的话核心也不能充分利用。

为了提高任务个数需要使用设置spark.streaming.concurrentJobs参数&＃xff1a;

spark-submit --conf spark.streaming.concurrentJobs&＃61;8 ....


实际上一个Job会分成多个Tasks&＃xff0c;每个CPU核心执行一个Task&＃xff0c;Task执行完成则Core被释放&＃xff0c;也就是说8个partitions的Streaming&＃xff0c;使用32个核心并不是只能执行4个Jobs&＃xff0c;可以根据Spark WebUI的executor页面核心的使用量&＃xff0c;适当的增大concurrentJobs或减少核心使用量。
像我这就是利用率比较低&＃xff0c;并且Streaming任务一直跟得上&＃xff0c;就可以适当降低Cores的数量。

2. GC优化
使用CMS内存收集器&＃xff1a;

spark-submit --conf "spark.executor.extraJavaOptions&＃61;-XX:&＃43;UseConcMarkSweepGC"


自从使用了这个收集器&＃xff0c;GC时间下来了&＃xff0c;内存也不容易超限了&＃xff0c;一口气上五楼都不费劲了&＃xff5e;具体原理不清楚&＃xff0c;回头补。

3. 有关cache
   数据有复用的位置一定要记得cache&＃xff0c;否则会从头开始执行处理流程。
   被cache的类型需要能够进行序列化。

4. 有关序列化与反序列化
   Driver会将Task的内容打包序列化发给Executor&＃xff0c;所以需要Task中所有被引用的类型都可以序列化。
   如果类型不可序列化则会报object not serializable的错误&＃xff0c;此时需要自己实现序列化与反序列化方法&＃xff0c;一般只需要实现反序列化方法&＃xff08;readObject&＃xff09;
   scala需要加上serializable注解&＃xff0c;java实现serializable接口。

private def readObject(in: ObjectInputStream):Unit &＃61; {//调用默认的ReadObject函数in.defaultReadObject()//重新初始化一些无法被序列化的内容this.init(this.config_map)}


对于无法序列化的属性&＃xff08;Mysql连接、Redis连接等等等)需要在属性前加上transient修饰符&＃xff0c;表示在序列化时忽略&＃xff0c;然后在readObject中再进行构造。

scala还有一个lazy修饰符&＃xff0c;表明使用时再进行构建&＃xff0c;所以也可以使用lazy&＃43;transient修饰符&＃xff0c;让其在使用时重新进行构建。

&＃64;transient lazy val logger:Logger &＃61; LogManager.getLogger(this.getClass.getName)


如下例子进行参考,一个包装的Kafka的类&＃xff1a;

class KafkaSink[K, V](createProducer: () &＃61;> KafkaProducer[K, V]) extends Serializable {/* This is the key idea that allows us to work around running intoNotSerializableExceptions. */lazy val producer &＃61; createProducer()def send(topic: String, key: K, value: V): Future[RecordMetadata] &＃61;producer.send(new ProducerRecord[K, V](topic, key, value))def send(topic: String, value: V): Future[RecordMetadata] &＃61;producer.send(new ProducerRecord[K, V](topic, value))
}object KafkaSink {import scala.collection.JavaConversions._def apply[K, V](config: Map[String, Object]): KafkaSink[K, V] &＃61; {val createProducerFunc &＃61; () &＃61;> {val producer &＃61; new KafkaProducer[K, V](config)sys.addShutdownHook {// Ensure that, on executor JVM shutdown, the Kafka producer sends// any buffered messages to Kafka before shutting down.producer.close()}producer}new KafkaSink(createProducerFunc)}def apply[K, V](config: java.util.Properties): KafkaSink[K, V] &＃61; apply(config.toMap)
}


5.广播变量
Task过程中使用的变量每次都会序列化传输一次&＃xff0c;如果想验证可以使用上面的方法重写readObject打印一些调试信息进行记录。

而一些长时间不变并且比较大、复杂的内容可以使用广播变量进行保存&＃xff0c;保证每个executor只存在一份该变量。

比如 Redis连接、MySQL连接、规则配置什么的就可以使用广播变量。

广播变量包裹的类同样需要能够序列化。

广播变量为只读变量。

详细广播变量的使用可以看如下文章&＃xff1a;
https://www.jianshu.com/p/3bd18acd2f7f

6. 利用广播变量进行配置更新&＃xff1a;

详细可以看这篇文章&＃xff1a;
https://www.cnblogs.com/liuliliuli2017/p/6782687.html
某个大佬写的包装类&＃xff1a;

// This wrapper lets us update brodcast variables within DStreams&＃39; foreachRDD
// without running into serialization issues
case class BroadcastWrapper[T: ClassTag](&＃64;transient private val ssc: StreamingContext,&＃64;transient private val _v: T) {&＃64;transient private var v &＃61; ssc.sparkContext.broadcast(_v)def update(newValue: T, blocking: Boolean &＃61; false): Unit &＃61; {// 删除RDD是否需要锁定v.unpersist(blocking)v &＃61; ssc.sparkContext.broadcast(newValue)}def value: T &＃61; v.valueprivate def writeObject(out: ObjectOutputStream): Unit &＃61; {out.writeObject(v)}private def readObject(in: ObjectInputStream): Unit &＃61; {v &＃61; in.readObject().asInstanceOf[Broadcast[T]]}
}


ElasticSearch

1. index名字只能小写&＃xff0c;赤裸裸的教训啊&＃61; &＃61;
2. 如果只使用单个ES账号&＃xff0c;可以使用全局配置的es.nodes等参数&＃xff0c;如果使用多个ES源的话可以写入多个配置MAP:

var esin_setting &＃61; Map[String,String]("es.nodes"->"es1","es.port"->"7001")var esout_setting &＃61; Map[String,String]("es.nodes"->"es2","es.port"->"7001","es.scroll.size"->"5000")
val rdd &＃61; sc.esRDD("indexin", query,esin_setting)
rdd.saveToEs("esout/type1",esout_setting)


3.更多复杂的配置项&＃xff0c;可以参考ElasticSearch的官网配置文档&＃xff1a;
https://www.elastic.co/guide/en/elasticsearch/hadoop/current/configuration.html