案例1.spark和flink分别实现作业配置动态更新案例

一、背景

在实时作业（如 Spark Streaming、Flink 等流处理作业）中，通过外部配置管理系统动态修改配置，有以下优点：

1. 无需重启作业，实现配置热更新
好处：实时作业通常需要长时间运行，重启会导致数据丢失或处理延迟。通过外部配置动态更新，可以在作业运行时修改配置（如并行度、窗口大小、超时时间等），而无需重启作业。

2. 灵活应对业务需求变化
好处：实时作业通常需要快速响应业务需求的变化（如规则调整、参数优化等）。通过外部配置管理，可以快速更新业务逻辑或参数，而无需重新部署代码。

3. 集中化管理配置
好处：将配置集中存储在外部的配置管理系统（如 ZooKeeper、Consul、Nacos、数据库等），便于统一管理和维护。多个作业可以共享同一份配置，避免配置分散和重复。

通常初始化sparkConf 作为 Spark 应用程序的配置对象，在运行时设置配置参数，但是大多数配置在 SparkContext 初始化后无法更改了。

如果想让程序在运行时做参数的动态调整，以下有两种思路可供参考，通过代码案例可以更深一步理解。

二、解决

spark%20broadcast%E5%B9%BF%E6%92%AD%E5%8F%98%E9%87%8F" name="1.%E6%96%B9%E6%B3%951%EF%BC%9Aspark%20broadcast%E5%B9%BF%E6%92%AD%E5%8F%98%E9%87%8F">1.方法1：spark broadcast广播变量

a. 思路

可以将数据字典或者规则放入文件当中，spark读取加载，并将这些字典配置广播发送到各个节点。

b. 案例

① 需求

根据ip地址查找其所属，实现根据ip对应所属的规则，将该规则广播出去，为数据中的ip附加所属。将ip与所属地址的规则broadcast广播到各个executor中。

② 数据

access.log
样例：字符分隔\t 分别是时间\tip\t网址

20161127172603 12.197.80.128 http://scala.bjut.com.cn/scala/course/8.html
20161127172605 128.44.80.128 http://java.bjut.com.cn/java/course/31.html
20161127172605 12.18.80.188 http://java.bjut.com.cn/java/course/22.html
20161127172610 197.160.85.168 http://java.bjut.com.cn/java/course/18.html

实现根据ip对应所属的规则，将该规则广播出去，为数据中的ip附加所属。实现代码如下：

③ 代码

IpLocation.scala

package com.spark.demo.broadcastimport org.apache.spark.SparkConf
import org.apache.spark.SparkContext/**
 * 根据ip地址查找其所属
 * 将ip与所属地址的规则broadcast广播到各个executor中
 */
object IpLocation {    def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setAppName("ip location").setMaster("local[2]")
        val sc = new SparkContext(conf)        val ipRulesRdd = sc.textFile("src\\com\\spark\\demo\\broadcast\\ip-by-country.csv").map { line => 
            val fields = line.split(",")
            val startIpNum = fields(2)
            val endIpNum = fields(3)
            val cityName = fields(5)
            (startIpNum, endIpNum, cityName)
        }
        //全部的ip映射规则
        val ipRulesArray = ipRulesRdd.collect //将数据收集到driver，为后面广播做准备。该变量值在driver中有，worker中不存在
        val bIpRules = sc.broadcast(ipRulesArray)        //加载要处理的数据
        val ipsRdd = sc.textFile("src\\com\\spark\\demo\\broadcast\\access.log").map { line =>  
            val fields = line.split("\t")
            val ip = fields(1)
            ip
        }        val result = ipsRdd.map { ip => 
            val ipNum = IpUtil.ip2Long(ip)
            val info = IpUtil.searchContentByKey(bIpRules.value, ipNum)
            val countryName = info.split(",")(2)
            (ip, countryName)
        }        result.collect().foreach(println)
        sc.stop()
    }}

IpUtil.scala

package com.spark.demo.broadcastimport scala.collection.mutable.ArrayBuffer
import scala.collection.mutable.HashMapobject IpUtil {
  /*
     * 将ip地址转换成数字long
     */
    def ip2Long(ip:String):Long = {
        val fragments = ip.split("[.]") 
        var ipNum = 0L
        for(i <- 0 until fragments.length) {
            ipNum = fragments(i).toLong | ipNum << 8L
        }
        ipNum
    }    def ip2Long2(ip:String) = {
        val fragments = ip.split("[.]")
        val ipNum = 16777216L*fragments(0).toLong+65536L*fragments(1).toLong+
                256L*fragments(2).toLong + fragments(3).toLong
        ipNum
    }
    /**
     * 将数字转换成ip地址
     */
    def long2Ip(ipNum:Long) = {
        val mask = List(0x000000FF,0x0000FF00,0x00FF0000,0xFF000000)
        val ipInfo = new StringBuffer
        var num = 0L
        for(i <- 0 until 4) {
            num = (ipNum & mask(i)) >> (i*8)
            if(i>0) ipInfo.insert(0, ".")