在大菠萝中遇到的一些小问题

发表于

大菠萝的任务

大菠萝全名diablo technology,是一家硬件技术外企.但是我是一名纯软界的小小码农,怎么会帮一个硬件公司做事呢.起初是这家公司中国区的技术人员跟我们组里的陆博大大比较熟,他们想招一两个懂Spark的人帮他们做一些测试和性能上的优化来得出他们研发的新硬件M1相比于市场上普通内存的特性与性能差异.于是陆博建议我担任这份兼职,工作不多而且还可以增一份不错的收入,众所周知,这种事情我是无法拒绝的.

为了推广M1,他们需要在M1上运行很多工业市场上流行的吃内存的软件与平台,包括spark,mysql,redis等.所以这三个月我就没事帮他们写一点测试的代码,优化spark的配置,寻找运行错误的bug,搭建redis-cluster这种活.期间爬过很多坑,有几个瞬间我感觉到自己是技术顾问,让我小小地膨胀一下.

C程序的测试

C语言我是有好几年没碰了,作为一个大四以后一直在jvm语言中游走,有时用python写点小项目的人,几乎对指针这种东西处于懵逼模式.

代码的编写和测试流程我就不详细说了,说说遇到的段错误解决方案.也是这次跑C程序让我有了了解这方面知识的契机.发生了段错误/core dump,总结一下原因,一般是这几个:

  • 访问了不存在的内存地址
  • 访问了系统保护的内存地址
  • 访问了只读的内存地址等等情况

先用gdb -g -o xx xx.c 生成gdb可调试的文件。说到gdb调试,大概要知道这几个命令:

  1. gdb xx 进入gdb调试界面

  2. b 行数 在第多少行打断点

  3. 按r进入运行状态

  4. n是单步调试

  5. s进入函数中调试

  6. q是退出

继续说解决core dump的bug, 然后运行程序, ./xx发生段错误后会生成core文件.接着运行命令gdb xx core.36129.输入where 会打印出具体的core dump发生在代码中哪个位置，方便定位bug。

还有一个bug是,每次都在内存地址被free的时候报无效指针,这种情况一般包含以下几个原因:

  1. 一个地址被free了两次, 当然这个很容易查出来.

  2. free的地址已经不是当时申请内存的起始地址了,注意可能在程序中的一些函数中对其做了一些操作.

Spark-Sql测试框架中踩得坑

大菠萝公司的喵叽(算我们的上司大人)想在机器上测试Spark Sql在M1上跑的性能，让我调研下Spark-sql-perf这个已有的测试框架(项目地址)。看了一下框架代码，差不多就是写了很多sql语句封装成一个个benchmark对象然后有一个多线程模式，然后一起开测。首先，在大菠萝的机器上部署测试框架，然后写一个简单的测试程序。代码如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
object SqlTest extends Serializable{
  def main(args: Array[String]): Unit = {
    val sparkConf  = new SparkConf().setAppName("test")
    .setMaster("local")
    val sc = new SparkContext(sparkConf)
    val sqlContext = new org.apache.spark.sql.SQLContext(sc)
   val tables = new Tables(sqlContext, "/Users/iceke/projects/tpcds-kit/tools", 
   Integer.parseInt("1"))
    tables.genData("data","parquet",true, false, false, false, false)
    val tableNames = Array("call_center", "catalog_page",  "catalog_returns", "catalog_sales",
      "customer", "customer_address",  "customer_demographics", "date_dim",
      "household_demographics", "income_band",   "inventory", "item", "promotion",
      "reason", "ship_mode",  "store", "store_returns",  "store_sales", "time_dim",
      "warehouse", "web_page",   "web_returns", "web_sales", "web_site")
    for(i <- 0 to tableNames.length - 1) {
      val a = sqlContext.read.parquet("data" + "/" + tableNames{i})
     // sc.broadcast(a)
      a.registerTempTable(tableNames{i})
    }
    val tpcds = new TPCDS (sqlContext = sqlContext)
    val experiment = tpcds.runExperiment(tpcds.tpcds1_4Queries, iterations = 1,forkThread=false)
    experiment.waitForFinish(60*60*10)
  }

先生成数据，大概几十张表。然后加载这几十张表，进行benchmark实验。本来过程很简单，但是需求是无止境的，喵叽说试试把所有的表cache到内存中，这样进行查询时快一点，听上去非常简单，在注册后加上一行cache表的代码就可以了。部分代码如下：

1
2
3
4
 val a = sqlContext.read.parquet("data" + "/" + tableNames{i})
// sc.broadcast(a)
 a.registerTempTable(tableNames{i})
 sqlContext.cacheTable(tableNames{i})

我在mac上用local模式测了一下ok便提交给喵叽。

事情永远不会像想象的那么顺利，喵叽在集群上一测，就通过了几个benchmark，然后大量的stage报错，全部都是failed to get broadcast(TorrentBroadcast)异常，然后stage 直接失败，查看executor的日志发现已经有的broacast被remove了，但是接下来的加个task又会去获取这些broadcast，便会直接失败。这明显是不符合逻辑的，明明cache了所有表，靠异常堆栈信息并不好定位到错误的地方。我看了spark主页发现storage页面cache的RDD过一段时间就会消失，这也是一个重要的线索。

测试了半天还是出错，我开始静下心慢慢跟着框架的代码走，我发现最后每一个线程都会执行一个doBenchmark方法，
方法里面最后执行完查询操作之后会做一些善后处理。相关代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
final def benchmark(
    includeBreakdown: Boolean,
    description: String = "",
    messages: ArrayBuffer[String],
    timeout: Long,
    forkThread: Boolean = true): BenchmarkResult = {
  logger.info(s"$this: benchmark")
  sparkContext.setJobDescription(s"Execution: $name, $description")
  beforeBenchmark()
  val result = if (forkThread) {
    runBenchmarkForked(includeBreakdown, description, messages, timeout)
  } else {
    doBenchmark(includeBreakdown, description, messages)
  }
  println("!!!!!!!!!!!!!!!!after Bench")
  afterBenchmark(sqlContext.sparkContext)
  result

这个善后处理是afterBenchmark方法，方法代码：

1
2
3
4
5
6
7
8
private def afterBenchmark(sc: SparkContext): Unit = {
  // Best-effort clean up of weakly referenced RDDs, shuffles, and broadcasts
  System.gc()
  // Remove any leftover blocks that still exist
  sc.getExecutorStorageStatus
      .flatMap { status => status.blocks.map { case (bid, _) => bid } }
      .foreach { bid => SparkEnv.get.blockManager.master.removeBlock(bid) }
}

它会对每个相关block的id进行移除，无论它是否做了cache。所以解决方法是注释掉afterBenchmark方法，与上次堆外内存的bug一样，发现bug的过程是痛苦的，解决方案是简单到发指的,让人痛苦,又让人快乐.

所以说，别人的框架不是万能的，虽然你操作起来更便捷，但你必须遵守它制定的那一些规则，有些规则是操蛋的，当你使用它时，如果一直被操蛋的bug所围困，就需要看看它的源码看看是否有问题，或者与你的策略存在偶然性的冲突。

最后, 我的梦想是成为规则的制定者.