<commit_before>/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.intel.analytics.bigdl.example

import com.intel.analytics.bigdl.dataset.image.RGBImage
import com.intel.analytics.bigdl.example.ImageNetUtils._
import com.intel.analytics.bigdl.example.Utils._
import com.intel.analytics.bigdl.models.imagenet.{GoogleNet_v1_NoAuxClassifier, GoogleNet_v2_NoAuxClassifier}
import com.intel.analytics.bigdl.nn._
import com.intel.analytics.bigdl._
import com.intel.analytics.bigdl.models.alexnet.AlexNet
import com.intel.analytics.bigdl.optim._
import com.intel.analytics.bigdl.optim.SGD.{EpochSchedule, EpochStep, Poly, Regime}
import com.intel.analytics.bigdl.parameters.{AllReduceParameterManager, ImprovedAllReduceParameterManager, OneReduceParameterManager}
import com.intel.analytics.bigdl.tensor._
import com.intel.analytics.bigdl.utils.{RandomGenerator, T}
import org.apache.hadoop.io.Text
import org.apache.log4j.{Level, Logger}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object ImageNetParallel {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.ERROR)
    Logger.getLogger("akka").setLevel(Level.ERROR)
    Logger.getLogger("breeze").setLevel(Level.ERROR)
    Logger.getLogger("com.intel.analytics.bigdl.optim").setLevel(Level.INFO)

    val parser = getParser()

    parser.parse(args, defaultParams).map { params => {
      train(params)
    }
    }.getOrElse {
      sys.exit(1)
    }
  }

  private def train(params: Params) = {
    val trainFiles = params.folder + "/train"
    val testFiles = params.folder + "/validation"
    val labelFile = params.labelsFile
    val partitionNum = params.partitionNum
    val optType = params.masterOptM
    val netType = params.net
    val classNum = params.classNum
    val conf = new SparkConf().setAppName(s"ImageNet class[Float]: ${params.classNum} " +
      s"Parallelism: ${params.parallelism.toString}, partition : ${params.partitionNum}, " +
      s"masterConfig: ${params.masterConfig}, workerConfig: ${params.workerConfig}")

    conf.setExecutorEnv("MKL_DISABLE_FAST_MM", "1")
    conf.setExecutorEnv("KMP_BLOCKTIME", "0")
    conf.setExecutorEnv("OMP_WAIT_POLICY", "passive")
    conf.setExecutorEnv("OMP_NUM_THREADS", "1")
    conf.set("spark.task.maxFailures", "1")
    conf.set("spark.shuffle.blockTransferService", "nio")
    conf.set("spark.akka.frameSize", "10") // akka networking speed is slow
    val sc = new SparkContext(conf)
    val cropImage = params.crop
    println(s"crop $cropImage")

    val labelsMap = getLabels(labelFile)
    val trainData = if (cropImage) {
      loadCroppedData(trainFiles, sc, labelsMap, classNum + 0.5)
    } else {
      loadData(trainFiles, sc, labelsMap, classNum + 0.5)
    }
    println(s"Count train data...")
    val totalSize = trainData.count()
    println(s"train data size is ${totalSize}")
    println("Estimate mean and std...")
    val sum = computeMean(trainData, params.crop)
    val meanR = sum._1.toFloat / totalSize
    val meanG = sum._2.toFloat / totalSize
    val meanB = sum._3.toFloat / totalSize
    val mean = (meanR, meanG, meanB)
    println(s"meanR is $meanR meanG is $meanG meanB is $meanB")
    val varSum = computeVar(trainData, meanR, meanG, meanB, params.crop)
    val varR = varSum._1.toFloat / totalSize
    val varG = varSum._2.toFloat / totalSize
    val varB = varSum._3.toFloat / totalSize
    val std = (varR, varG, varB)
    println(s"varR is $varR varG is $varG varB is $varB")

    val criterion = ClassNLLCriterion[Float]()
    val model = netType match {
      case "alexnet" => AlexNet(classNum)
      case "googlenet" => GoogleNet.getModelCaffe[Float](classNum)
      case "googlenet_v2" => GoogleNet_v2_NoAuxClassifier(classNum)
      case "googlenet_v1" => GoogleNet_v1_NoAuxClassifier(classNum)
    }
    println(model)

    val driverConfig = params.masterConfig.clone()
    val workerConfig = params.workerConfig.clone()
    workerConfig("profile") = true

    netType match {
      case "googlenet" => driverConfig("learningRateSchedule") = Poly(0.5, 62000)
      case "googlenet_v2" =>
      driverConfig("learningRateSchedule") = EpochSchedule(Array(
        Regime(1, 18, T("learningRate" -> 1e-2, "weightDecay" -> 5e-4)),
        Regime(19, 29, T("learningRate" -> 5e-3, "weightDecay" -> 5e-4)),
        Regime(30, 43, T("learningRate" -> 1e-3, "weightDecay" -> 0.0)),
        Regime(44, 52, T("learningRate" -> 5e-4, "weightDecay" -> 0.0)),
        Regime(53, 1000, T("learningRate" -> 1e-4, "weightDecay" -> 0.0))
      ))
      case "googlenet_v1" =>
        driverConfig("learningRateSchedule") = Poly(0.5, 62000)
    }
    // driverConfig("learningRateSchedule") = Poly(0.5, 62000)
    // driverConfig("learningRateSchedule") = Poly(0.5, 90000)
    // driverConfig("learningRateSchedule") = EpochStep(2, 0.96)
    // driverConfig("learningRateSchedule") = EpochSchedule(Array(
    //   Regime(1, 18, T("learningRate" -> 1e-2, "weightDecay" -> 5e-4)),
    //   Regime(19, 29, T("learningRate" -> 5e-3, "weightDecay" -> 5e-4)),
    //   Regime(30, 43, T("learningRate" -> 1e-3, "weightDecay" -> 0.0)),
    //   Regime(44, 52, T("learningRate" -> 5e-4, "weightDecay" -> 0.0)),
    //   Regime(53, 1000, T("learningRate" -> 1e-4, "weightDecay" -> 0.0))
    // ))
<commit_msg>changeback googlenetv1 model<commit_after>/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.intel.analytics.bigdl.example

import com.intel.analytics.bigdl.dataset.image.RGBImage
import com.intel.analytics.bigdl.example.ImageNetUtils._
import com.intel.analytics.bigdl.example.Utils._
import com.intel.analytics.bigdl.models.imagenet.{GoogleNet_v1_NoAuxClassifier, GoogleNet_v2_NoAuxClassifier}
import com.intel.analytics.bigdl.nn._
import com.intel.analytics.bigdl._
import com.intel.analytics.bigdl.models.alexnet.AlexNet
import com.intel.analytics.bigdl.models.googlenet.{GoogleNet_v1, GoogleNet_v2}
import com.intel.analytics.bigdl.optim._
import com.intel.analytics.bigdl.optim.SGD.{EpochSchedule, EpochStep, Poly, Regime}
import com.intel.analytics.bigdl.parameters.{AllReduceParameterManager, ImprovedAllReduceParameterManager, OneReduceParameterManager}
import com.intel.analytics.bigdl.tensor._
import com.intel.analytics.bigdl.utils.{RandomGenerator, T}
import org.apache.hadoop.io.Text
import org.apache.log4j.{Level, Logger}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object ImageNetParallel {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.ERROR)
    Logger.getLogger("akka").setLevel(Level.ERROR)
    Logger.getLogger("breeze").setLevel(Level.ERROR)
    Logger.getLogger("com.intel.analytics.bigdl.optim").setLevel(Level.INFO)

    val parser = getParser()

    parser.parse(args, defaultParams).map { params => {
      train(params)
    }
    }.getOrElse {
      sys.exit(1)
    }
  }

  private def train(params: Params) = {
    val trainFiles = params.folder + "/train"
    val testFiles = params.folder + "/validation"
    val labelFile = params.labelsFile
    val partitionNum = params.partitionNum
    val optType = params.masterOptM
    val netType = params.net
    val classNum = params.classNum
    val conf = new SparkConf().setAppName(s"ImageNet class[Float]: ${params.classNum} " +
      s"Parallelism: ${params.parallelism.toString}, partition : ${params.partitionNum}, " +
      s"masterConfig: ${params.masterConfig}, workerConfig: ${params.workerConfig}")

    conf.setExecutorEnv("MKL_DISABLE_FAST_MM", "1")
    conf.setExecutorEnv("KMP_BLOCKTIME", "0")
    conf.setExecutorEnv("OMP_WAIT_POLICY", "passive")
    conf.setExecutorEnv("OMP_NUM_THREADS", "1")
    conf.set("spark.task.maxFailures", "1")
    conf.set("spark.shuffle.blockTransferService", "nio")
    conf.set("spark.akka.frameSize", "10") // akka networking speed is slow
    val sc = new SparkContext(conf)
    val cropImage = params.crop
    println(s"crop $cropImage")

    val labelsMap = getLabels(labelFile)
    val trainData = if (cropImage) {
      loadCroppedData(trainFiles, sc, labelsMap, classNum + 0.5)
    } else {
      loadData(trainFiles, sc, labelsMap, classNum + 0.5)
    }
    println(s"Count train data...")
    val totalSize = trainData.count()
    println(s"train data size is ${totalSize}")
    println("Estimate mean and std...")
    val sum = computeMean(trainData, params.crop)
    val meanR = sum._1.toFloat / totalSize
    val meanG = sum._2.toFloat / totalSize
    val meanB = sum._3.toFloat / totalSize
    val mean = (meanR, meanG, meanB)
    println(s"meanR is $meanR meanG is $meanG meanB is $meanB")
    val varSum = computeVar(trainData, meanR, meanG, meanB, params.crop)
    val varR = varSum._1.toFloat / totalSize
    val varG = varSum._2.toFloat / totalSize
    val varB = varSum._3.toFloat / totalSize
    val std = (varR, varG, varB)
    println(s"varR is $varR varG is $varG varB is $varB")

    val criterion = ClassNLLCriterion[Float]()
    val model = netType match {
      case "alexnet" => AlexNet(classNum)
      case "googlenet" => GoogleNet.getModelCaffe[Float](classNum)
      case "googlenet_v2" => GoogleNet_v2(classNum)
      case "googlenet_v1" => GoogleNet_v1(classNum)
    }
    println(model)

    val driverConfig = params.masterConfig.clone()
    val workerConfig = params.workerConfig.clone()
    workerConfig("profile") = true

    netType match {
      case "googlenet" => driverConfig("learningRateSchedule") = Poly(0.5, 62000)
      case "googlenet_v2" =>
      driverConfig("learningRateSchedule") = EpochSchedule(Array(
        Regime(1, 18, T("learningRate" -> 1e-2, "weightDecay" -> 5e-4)),
        Regime(19, 29, T("learningRate" -> 5e-3, "weightDecay" -> 5e-4)),
        Regime(30, 43, T("learningRate" -> 1e-3, "weightDecay" -> 0.0)),
        Regime(44, 52, T("learningRate" -> 5e-4, "weightDecay" -> 0.0)),
        Regime(53, 1000, T("learningRate" -> 1e-4, "weightDecay" -> 0.0))
      ))
      case "googlenet_v1" =>
        driverConfig("learningRateSchedule") = Poly(0.5, 62000)
    }
    // driverConfig("learningRateSchedule") = Poly(0.5, 62000)
    // driverConfig("learningRateSchedule") = Poly(0.5, 90000)
    // driverConfig("learningRateSchedule") = EpochStep(2, 0.96)
    // driverConfig("learningRateSchedule") = EpochSchedule(Array(
    //   Regime(1, 18, T("learningRate" -> 1e-2, "weightDecay" -> 5e-4)),
    //   Regime(19, 29, T("learningRate" -> 5e-3, "weightDecay" -> 5e-4)),
    //   Regime(30, 43, T("learningRate" -> 1e-3, "weightDecay" -> 0.0)),
    //   Regime(44, 52, T("learningRate" -> 5e-4, "weightDecay" -> 0.0)),
    //   Regime(53, 1000, T("learningRate" -> 1e-4, "weightDecay" -> 0.0))
    // ))

{'fraction_non_alphanumeric': 0.13304404781398396, 'fraction_numerical': 0.02382511871622728, 'mean_word_length': 3.7492223950233283, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 59, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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}}
{'dir': 'git-commits-cleaned', 'id': '6649533', 'n_tokens_mistral': 4228, 'n_tokens_neox': 4012, 'n_words': 915}