<jupyter_start><jupyter_text>Load Bands From HDFS into RDD<jupyter_code>//rdd in the form year; PortableDataStream
// B0 (BLUE, 0.46 µm), B2 (RED, 0.68 µm), B3 (NIR,0.85 µm)
var rdd_blue = sc.binaryFiles("hdfs:///user/hadoop/TOC_Cropped/B0/*.img").map(m => (FilenameUtils.getBaseName(m._1).split("_")(3).slice(0,8), m._2))
var rdd_red = sc.binaryFiles("hdfs:///user/hadoop/TOC_Cropped/B2/*.img").map(m => (FilenameUtils.getBaseName(m._1).split("_")(3).slice(0,8), m._2))
var rdd_nir = sc.binaryFiles("hdfs:///user/hadoop/TOC_Cropped/B3/*.img").map(m => (FilenameUtils.getBaseName(m._1).split("_")(3).slice(0,8), m._2))
var rdd_sm = sc.binaryFiles("hdfs:///user/hadoop/TOC_Cropped/SM/*.img").map(m => (FilenameUtils.getBaseName(m._1).split("_")(3).slice(0,8), m._2))

var all_bands :RDD[(String,PortableDataStream,PortableDataStream,PortableDataStream,PortableDataStream)]  = rdd_blue.join(rdd_red).join(rdd_nir).join(rdd_sm).map(x => (x._1, x._2._1._1._1,  x._2._1._1._2, x._2._1._2 , x._2._2))


print(all_bands.count)<jupyter_output><empty_output><jupyter_text>Derive VIs (NDVI or EVI)<jupyter_code>def calculateEVI( blue_stream: PortableDataStream, red_stream: PortableDataStream, nir_stream: PortableDataStream, sm_stream: PortableDataStream  ): Array[Float] ={
    var blue_array : Array[Byte] = blue_stream.toArray()
    
    var evi = new Array[Float](blue_array.length/2)
    
    var  blue = ByteBuffer.wrap(blue_array).order(ByteOrder.LITTLE_ENDIAN)
    var  red = ByteBuffer.wrap(red_stream.toArray()).order(ByteOrder.LITTLE_ENDIAN)
    var  nir = ByteBuffer.wrap(nir_stream.toArray()).order(ByteOrder.LITTLE_ENDIAN)
    var  sm = ByteBuffer.wrap(sm_stream.toArray()).order(ByteOrder.LITTLE_ENDIAN)
   
    //loop on the pixel positions 
    for(i <- 0 until blue_array.length/2 ){ 

        var byte_quality: Array[Byte] = Array[Byte] (sm.get) //get will increment the buffer
        var byte_qualit2: Array[Byte] = Array[Byte] (sm.get) //get will increment the buffer
        var bitset: BitSet = BitSet.valueOf(byte_quality)
        var bitse2: BitSet = BitSet.valueOf(byte_qualit2)
        var blue_value = blue.getShort *  0.0005
        var red_value = red.getShort *  0.0005
        var nir_value = nir.getShort *  0.0005
        
        if(bitset.get(0) == false  &&  bitset.get(1) == false && bitset.get(6) == false && bitset.get(5) == false && bitset.get(7) == false && bitse2.get(0) == false){
        
            //Appy formulas: Commented code is the EVI VIs
            var denominator = nir_value + red_value // NDVI
           // var denominator = (nir_value + 6.0*red_value - 7.5*blue_value) + 1.0 ///EVI

            if(denominator != 0  ){
                  evi(i) = ((nir_value - red_value)/denominator).toFloat // NDVI
//               evi(i)= (2.5*(nir_value - red_value)/denominator).toFloat     //EVI
                 
            }else{
                evi(i)= -5f //error Code
            }
        }
        else{
            evi(i)= -6f //error Code
        }
    }
    return evi
}
       


var result :RDD[(String, Array[Float])] = all_bands.map(x => (x._1,calculateEVI(x._2,x._3,x._4,x._5)))
result.take(1).foreach(arr => println(arr._1,arr._2.mkString(", ")))<jupyter_output><empty_output><jupyter_text>Save Results<jupyter_code>def save (timestamp: String, data: Array[Float]): Integer = {

    var byteBuffer: ByteBuffer = ByteBuffer.allocateDirect(4*data.length).order(ByteOrder.LITTLE_ENDIAN)
    //The FloatBuffer is backed by the byteBuffer (same data)
    var buffer: FloatBuffer = byteBuffer.asFloatBuffer.put(FloatBuffer.wrap(data))
    
    var file = new File("/data/local/home/parrot/minio2/ndvi_data_new/" + timestamp + ".img")

   //overwrite the file
    var channel = new FileOutputStream(file, false).getChannel()
    channel.write(byteBuffer)
    channel.close
    return 1
}<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.15662962005686223, 'fraction_numerical': 0.02558800723701215, 'mean_word_length': 3.3240223463687153, 'pattern_counts': {'":': 0, '<': 14, '<?xml version=': 0, '>': 20, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '3', 'dir': 'jupyter-structured-clean-dedup', 'id': '588092', 'license': 'Apache-2.0', 'n_tokens_mistral': 1485, 'n_tokens_neox': 1418, 'n_words': 241, 'path': 'VIs_calculations.ipynb', 'repo_name': 'phenology/platform-for-pheno-products', 'size': '3869'}