在spark中使用for循环会不会对性能造成影响

Question

在spark中使用for循环会不会对性能造成影响

Answer 1

Michael_PK 2020-08-11 17:20:16

举个例子看看

• 提问者 孤独观测者 #1

  hbaseRDD.filter(x =>{
        val time = Bytes.toString(x._2.getValue(Bytes.toBytes("info"), Bytes.toBytes("time")))
        time > "2020-08-11T00:00:00"
      }).map(x => {
        // 根据mac来对数据进行分组
        val id = Bytes.toString(x._2.getValue(Bytes.toBytes("info"), Bytes.toBytes("id")))
        val mac = Bytes.toString(x._2.getValue(Bytes.toBytes("info"), Bytes.toBytes("mac")))
        val time = Bytes.toString(x._2.getValue(Bytes.toBytes("info"), Bytes.toBytes("time")))
        val companyId = Bytes.toString(x._2.getValue(Bytes.toBytes("info"), Bytes.toBytes("companyId")))
  
        (mac, Bike(mac, time, id, companyId))
      }).groupByKey().collect().map(x => {
        //得到同一单车mac下所有的Bike
        val mac = x._1
        val value:Iterable[Bike] = x._2
        var list:ListBuffer[Bike] = new ListBuffer[Bike]()
  
        var tmp:String = value.iterator.next().id
        list+=value.iterator.next()
        println("单车："+mac)
  
        for (elem <- value) {
  
          println("    时间： "+ elem.time+",单车轨迹"+elem.id)
          if(elem.id != tmp){
            list+=elem
            tmp = elem.id
          }
        }
        println("测试："+ list)
        list
      })

• 提问者 孤独观测者 #2

  想根据Bike这个对象中的id（id代表一个地点），对hbase中获取的数据进行去重。现在可能在一个地点半小时内上传了多个数据，但我只想在一个地点保留一个数据

• Michael_PK 回复 提问者 孤独观测者 #3

  groupByKey().collect() 这种方法使用的时候要注意，因为collect是将所有数据返回到driver的，你driver是否能抗的住，这是个问题

Answer 2

提问者 孤独观测者 2020-08-11 22:34:34

hbaseRDD.filter(x =>{
 val time = Bytes.toString(x._2.getValue(Bytes.toBytes("info"), Bytes.toBytes("time")))
 time > "2020-08-11T00:00:00"
}).map(x => {
 // 根据mac来对数据进行分组
 val id = Bytes.toString(x._2.getValue(Bytes.toBytes("info"), Bytes.toBytes("id")))
 val mac = Bytes.toString(x._2.getValue(Bytes.toBytes("info"), Bytes.toBytes("mac")))
 val time = Bytes.toString(x._2.getValue(Bytes.toBytes("info"), Bytes.toBytes("time")))
 val companyId = Bytes.toString(x._2.getValue(Bytes.toBytes("info"), Bytes.toBytes("companyId")))

 (mac, Bike(mac, time, id, companyId))
}).groupByKey().collect().map(x => {
 //得到同一单车mac下所有的Bike
 val mac = x._1
 val value:Iterable[Bike] = x._2
 var list:ListBuffer[Bike] = new ListBuffer[Bike]()

 var tmp:String = value.iterator.next().id
 list+=value.iterator.next()
 println("单车："+mac)

 for (elem <- value) {

   println("    时间： "+ elem.time+",单车轨迹"+elem.id)
   if(elem.id != tmp){
     list+=elem
     tmp = elem.id
   }
 }
 println("测试："+ list)
 list
})

• Michael_PK #1

  这种结果是聚合后的，没啥大问题的，因为聚合后的结果的数量级已经没多大了

• 提问者 孤独观测者 回复 Michael_PK #2

  我还是有些疑惑，可能因为java写多了。所以在spark里写for循环来处理数据，就感觉会不会比spark自带的算子效率低下？

• Michael_PK 回复 提问者 孤独观测者 #3

  能用算子的毋庸置疑，因为分布式

点开查看后面1条评论