Add SnakeCaseColumnNames class to custom functions

src/main/scala/com/yotpo/metorikku/code/steps/SnakeCaseColumnNames.scala

@@ -0,0 +1,23 @@
+package com.yotpo.metorikku.code.steps
+
+import com.yotpo.metorikku.exceptions.MetorikkuException
+
+object SnakeCaseColumnNames {
+  val message = "You need to send 1 parameters with the names of the table to change: table"
+
+  def run(ss: org.apache.spark.sql.SparkSession, metricName: String, dataFrameName: String, params: Option[Map[String, String]]): Unit = {
+    params match {
+      case Some(parameters) =>
+        val table = parameters("table")
+        val df = ss.table(table)
+        val snakeCaseColumnNames = df.columns.map(toSnake)
+        df.toDF(snakeCaseColumnNames: _*).createOrReplaceTempView(dataFrameName)
+      case None => throw MetorikkuException(message)
+    }
+  }
+
+  private def toSnake(name: String): String = "([a-z])([A-Z]+)".r.replaceAllIn(name, { m =>
+    m.group(1) + "_" + m.group(2)
+  }.toLowerCase())
+}
+

src/test/scala/com/yotpo/metorikku/code/steps/test/SnakeCaseColumnNamesTest.scala

@@ -0,0 +1,42 @@
+package com.yotpo.metorikku.code.steps.test
+
+import com.holdenkarau.spark.testing.DataFrameSuiteBase
+import com.yotpo.metorikku.code.steps.SnakeCaseColumnNames
+import org.apache.log4j.{Level, LogManager, Logger}
+import org.apache.spark.sql.{SQLContext, SparkSession}
+import org.scalatest.{FunSuite, _}
+
+//noinspection ScalaStyle
+class SnakeCaseColumnNamesTest extends FunSuite with DataFrameSuiteBase with BeforeAndAfterEach {
+  private val log: Logger = LogManager.getLogger(this.getClass)
+  private var sparkSession: SparkSession = _
+  Logger.getLogger("org").setLevel(Level.WARN)
+
+  override def beforeEach() {
+    sparkSession = SparkSession.builder().appName("udf tests")
+      .master("local")
+      .config("", "")
+      .getOrCreate()
+  }
+
+  test("SnakeCaseColumnNames") {
+    val sparkSession = SparkSession.builder.appName("test").getOrCreate()
+    val sqlContext = new SQLContext(sparkSession.sparkContext)
+    import sqlContext.implicits._
+
+    val employeeData = Seq(
+      ("James", 1),
+      ("Maria", 2)
+    )
+    employeeData.toDF("EmployeeNAme", "EMPLOYEESALARY").createOrReplaceTempView("employeeDataActual")
+    employeeData.toDF("employeeName", "employeeSalary").createOrReplaceTempView("employeeDataExpected")
+    SnakeCaseColumnNames.run(sparkSession, "", "employeeDataExpectedResult", Some(Map("table" -> "employeeDataActual")))
+
+    assertDataFrameEquals(sparkSession.table("employeeDataExpected"), sparkSession.table("employeeDataExpectedResult"))
+
+  }
+
+  override def afterEach() {
+    sparkSession.stop()
+  }
+}