为什么使用广播会增加时间？

Question

case class IpDB(startIp: Long, endIp: Long, company: String)
def ipMap(line: Array[String]):
    val Array(startIp, endIp, company) = line
    IpDB(startIp.toLong, endIp.toLong, company)
// The ip-db is just 300M in raw format.
// format of ip-data from s3:
// 100000 200000 GOOGLE
val ipData = sc.broadcast(sc.textFile("s3://....").map(_.split("\t", -1)).map(ipMap).collect().sortWith(_.startIp < _.startIp)).value

val dataA = sc.textFile("s3://...").map((ip, 1)).reduceByKey(_+_)
val dataB = sc.textFile("s3://...").map((ip, 1)).reduceByKey(_+_)

// will do ip company lookup here
dataA.fullOuterJoin(dataB).map(doIpCompanyLookUp(ipData, _)).collect()

代码只是从输入源获取ip数据，然后在加入后找到它的公司。以下是我的问题：

这段代码在生产中会运行2-3分钟，但当删除广播数据（只需在两个数据上连接）时，只需花费不到1分钟。当我查看spark的ui时，我发现gc时间可能是问题所在。

下面是运行此作业的设置：

spark-submit --master yarn --deploy-mode client --driver-memory 4g --num-executors 10 --executor-memory 8800m --executor-cores 4 --class ... XX.jar 

spark version: 1.6.1
10 m3.xlarge.

null

case class IpDB(startIp: Long, endIp: Long, company: String)
def ipMap(line: Array[String]):
    val Array(startIp, endIp, company) = line
    IpDB(startIp.toLong, endIp.toLong, company)
// The ip-db is just 300M in raw format.
// format of ip-data from s3:
// 100000 200000 GOOGLE
val dataA = sc.textFile("s3://...").map((ip, 1)).reduceByKey(_+_)
val dataB = sc.textFile("s3://...").map((ip, 1)).reduceByKey(_+_)

// will do ip company lookup here
val joinResult = dataA.fullOuterJoin(dataB)
val ipData = sc.broadcast(sc.textFile("s3://....").map(_.split("\t", -1)).map(ipMap).collect().sortWith(_.startIp < _.startIp)).value
joinResult.map(doIpCompanyLookUp(ipData, _)).collect()

更新2。由于生产代码相当复杂，与上面的伪代码不同，在代码顺序上稍作改变后，程序运行得更快，但我不确定问题的关键是初始化广播数据的位置。

Answer 1

没有太多考虑代码的情况下，行：

val ipData = sc.broadcast(sc.textFile("s3://....").map(_.split("\t", -1)).map(ipMap).collect().sortWith(_.startIp < _.startIp)).value

让我很担心。

您使用sc.textfile构建分布式RDD，只是为了通过collect()使其成为本地（驱动程序），然后通过sc.broadcast（！）使其再次分布式并可供执行程序使用正如您所看到的，有大量的数据来回发送。

sc.textFile("s3://....").map(_.split("\t", -1)).map(ipMap).cache