From 656ba41ddf88f8da6fbca28d4a116b0c4a58fff1 Mon Sep 17 00:00:00 2001
From: swasti <skakker@linekdin.com>
Date: Sun, 8 Oct 2017 19:15:15 +0530
Subject: [PATCH] Changed default Compression Codec from .snappy to .lz4 and
 changed the spark fetcher

---
 app-conf/FetcherConf.xml                         |  2 +-
 .../spark/SparkMetricsAggregator.scala           |  3 +--
 .../linkedin/drelephant/util/SparkUtils.scala    |  2 +-
 .../drelephant/util/SparkUtilsTest.scala         | 16 ++++++++--------
 4 files changed, 11 insertions(+), 12 deletions(-)

diff --git a/app-conf/FetcherConf.xml b/app-conf/FetcherConf.xml
index 4e9f14df8..a6789caa3 100644
--- a/app-conf/FetcherConf.xml
+++ b/app-conf/FetcherConf.xml
@@ -79,7 +79,7 @@
   -->
   <fetcher>
     <applicationtype>spark</applicationtype>
-    <classname>com.linkedin.drelephant.spark.fetchers.FSFetcher</classname>
+    <classname>com.linkedin.drelephant.spark.fetchers.SparkFetcher</classname>
   </fetcher>
 
   <!--
diff --git a/app/com/linkedin/drelephant/spark/SparkMetricsAggregator.scala b/app/com/linkedin/drelephant/spark/SparkMetricsAggregator.scala
index 70172a21e..c8216cdf4 100644
--- a/app/com/linkedin/drelephant/spark/SparkMetricsAggregator.scala
+++ b/app/com/linkedin/drelephant/spark/SparkMetricsAggregator.scala
@@ -29,8 +29,7 @@ import scala.util.Try
 
 
 class SparkMetricsAggregator(private val aggregatorConfigurationData: AggregatorConfigurationData)
-  extends HadoopMetricsAggregator {
-
+    extends HadoopMetricsAggregator {
   import SparkMetricsAggregator._
 
   private val logger: Logger = Logger.getLogger(classOf[SparkMetricsAggregator])
diff --git a/app/com/linkedin/drelephant/util/SparkUtils.scala b/app/com/linkedin/drelephant/util/SparkUtils.scala
index e7efd9d84..bcb3f3f43 100644
--- a/app/com/linkedin/drelephant/util/SparkUtils.scala
+++ b/app/com/linkedin/drelephant/util/SparkUtils.scala
@@ -180,7 +180,7 @@ trait SparkUtils {
   }
 
   private val IN_PROGRESS = ".inprogress"
-  private val DEFAULT_COMPRESSION_CODEC = "snappy"
+  private val DEFAULT_COMPRESSION_CODEC = "lz4"
 
   private val compressionCodecClassNamesByShortName = Map(
     "lz4" -> classOf[LZ4CompressionCodec].getName,
diff --git a/test/com/linkedin/drelephant/util/SparkUtilsTest.scala b/test/com/linkedin/drelephant/util/SparkUtilsTest.scala
index 632b49536..aee8fb850 100644
--- a/test/com/linkedin/drelephant/util/SparkUtilsTest.scala
+++ b/test/com/linkedin/drelephant/util/SparkUtilsTest.scala
@@ -25,7 +25,7 @@ import org.apache.hadoop.fs.{FSDataInputStream, FileStatus, FileSystem, Path, Pa
 import org.apache.hadoop.io.compress.CompressionInputStream
 import org.apache.log4j.Logger
 import org.apache.spark.SparkConf
-import org.apache.spark.io.SnappyCompressionCodec
+import org.apache.spark.io.{LZ4CompressionCodec, SnappyCompressionCodec}
 import org.mockito.BDDMockito
 import org.mockito.Matchers
 import org.scalatest.{FunSpec, Matchers, OptionValues}
@@ -46,8 +46,8 @@ class SparkUtilsTest extends FunSpec with org.scalatest.Matchers with OptionValu
         }
 
         val (fs, path) = sparkUtils.fileSystemAndPathForEventLogDir(hadoopConfiguration,
-              sparkConf,
-              Some("webhdfs://nn1.grid.example.com:50070/logs/spark"))
+          sparkConf,
+          Some("webhdfs://nn1.grid.example.com:50070/logs/spark"))
         fs.getUri.toString should be("webhdfs://nn1.grid.example.com:50070")
         path should be(new Path("/logs/spark"))
       }
@@ -180,7 +180,7 @@ class SparkUtilsTest extends FunSpec with org.scalatest.Matchers with OptionValu
         val sparkUtils = SparkUtilsTest.newFakeSparkUtilsForEventLog(
           new URI("webhdfs://nn1.grid.example.com:50070"),
           new Path("/logs/spark"),
-          new Path("application_1_1.snappy"),
+          new Path("application_1_1.lz4"),
           Array.empty[Byte]
         )
 
@@ -189,8 +189,8 @@ class SparkUtilsTest extends FunSpec with org.scalatest.Matchers with OptionValu
         val (path, codec) =
           sparkUtils.pathAndCodecforEventLog(sparkConf: SparkConf, fs: FileSystem, basePath: Path, "application_1", Some("1"))
 
-        path should be(new Path("webhdfs://nn1.grid.example.com:50070/logs/spark/application_1_1.snappy"))
-        codec.value should be(a[SnappyCompressionCodec])
+        path should be(new Path("webhdfs://nn1.grid.example.com:50070/logs/spark/application_1_1.lz4"))
+        codec.value should be(a[LZ4CompressionCodec])
       }
       it("returns the path and codec for the event log, given the base path and appid. Extracts attempt and codec from path") {
         val hadoopConfiguration = new Configuration(false)
@@ -300,8 +300,8 @@ object SparkUtilsTest extends MockitoSugar {
       BDDMockito.given(fs.exists(expectedPath)).willReturn(true)
       BDDMockito.given(fs.getFileStatus(expectedPath)).willReturn(expectedFileStatus)
       BDDMockito.given(fs.listStatus(org.mockito.Matchers.refEq(new Path( new Path(fileSystemUri), basePath)),
-                                      org.mockito.Matchers.any(filter.getClass))).
-                 willReturn(expectedStatusArray)
+        org.mockito.Matchers.any(filter.getClass))).
+        willReturn(expectedStatusArray)
       BDDMockito.given(fs.open(expectedPath)).willReturn(
         new FSDataInputStream(new FakeCompressionInputStream(new ByteArrayInputStream(bytes)))
       )