[Spark] Matrix의 Row를 전체 Vertex 개수로 두고 Col 기준으로 나누기 ( 중복 허용 )

• Matrix의 Row를 전체 Vertex 개수로 두고 Col 기준으로 나누기 ( 중복 허용 )

  • 예) Matrix의 전체 Column size가 9이고 Slice 단위를 3으로 한다면 Column은 다음과 같이 분할 됨.

  • 1-3, 2-4, 3-5, 4-6, 5-7, 6-8, 7-9

  • 예제코드

    • 1024 개의 열을 1000 개씩 중복 허용하며 Slice 하므로 총 25개의 Slice가 나온다.

    • 1-1000, 2-1001, 3-1002, ... , 25-1024

        import scala.math.BigDecimal
      
        val rdd = sc.textFile("s3://square-matrix/M_1024_1024_0.001.txt")
      
        // 입력 데이터 row, col 크기
        val row = 1024
        val col = 1024
      
        // 원하는 slice 단위
        val slice_unit = 1000
      
        // slice의 시작,끝 idx
        var start = 1
        var end = slice_unit
      
        for (start <- 1 to col-slice_unit+1){
            end = start + slice_unit - 1
      
            // col을 slice 단위로 자른다.
            val slice = rdd.map{x=>x.split(" ")}.map{x => (x(0).toInt, x(1).toInt, x(2).toDouble)}.filter{x=> x._2 >= start && x._2 <= end}
      
            // slice의 nnz    
            val nnz = slice.count
      
            // slice의 density를 소수점 6자리로 반올림
            val density = BigDecimal(nnz.toDouble / (row * slice_unit).toDouble).setScale(6, BigDecimal.RoundingMode.HALF_UP)
      
            val result = slice.map{x=>x._1 + " " + x._2 + " " + x._3}
      
            // s3에 동일한 파일이 있을 경우엔 저장하지 않는다.
              try{
                  result.coalesce(1).saveAsTextFile("s3://non-square-matrix/M_"+row+"_"+slice_unit+"_"+nnz+"_"+density+".txt")
                  println("s3://non-square-matrix/M_"+row+"_"+slice_unit+"_"+nnz+"_"+density+".txt")
              }catch{
                  case x: org.apache.hadoop.mapred.FileAlreadyExistsException   => 
                      { 
                          println("s3://non-square-matrix/M_"+row+"_"+slice_unit+"_"+nnz+"_"+density+".txt"+" Already Exists in S3")
                      }
              }
        }

• Matrix의 Col을 전체 Vertex 개수로 두고 Row 기준으로 나누기 ( 중복 허용 )

  • 예제 코드

      import scala.math.BigDecimal
    
      val rdd = sc.textFile("s3://square-matrix/M_1024_1024_0.001.txt")
    
      // 입력 데이터 row, col 크기
      val row = 1024
      val col = 1024
    
      // 원하는 slice 단위
      val slice_unit = 1000
    
      // slice의 시작,끝 idx
      var start = 1
      var end = slice_unit
    
      for (start <- 1 to col-slice_unit+1){
          end = start + slice_unit - 1
    
          // row을 slice 단위로 자른다.
          val slice = rdd.map{x=>x.split(" ")}.map{x => (x(0).toInt, x(1).toInt, x(2).toDouble)}.filter{x=> x._1 >= start && x._1 <= end}
    
          // slice의 nnz    
          val nnz = slice.count
    
          // slice의 density를 소수점 6자리로 반올림
          val density = BigDecimal(nnz.toDouble / (row * slice_unit).toDouble).setScale(6, BigDecimal.RoundingMode.HALF_UP)
    
          val result = slice.map{x=>x._1 + " " + x._2 + " " + x._3}
    
          // s3에 동일한 파일이 있을 경우엔 저장하지 않는다.
            try{
                result.coalesce(1).saveAsTextFile("s3://non-square-matrix/M_"+slice_unit+"_"+col+"_"+nnz+"_"+density+".txt")
                println("s3://non-square-matrix/M_"+slice_unit+"_"+col+"_"+nnz+"_"+density+".txt"+" is put in S3")
            }catch{
                case x: org.apache.hadoop.mapred.FileAlreadyExistsException   => 
                    { 
                        println("s3://non-square-matrix/M_"+slice_unit+"_"+col+"_"+nnz+"_"+density+".txt"+" Already Exists in S3")
                    }
            }
      }