Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Handle the case where we don't have event log compression enabled and… #372

Open
wants to merge 2 commits into
base: master
Choose a base branch
from
Open

Handle the case where we don't have event log compression enabled and… #372

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a6a96b8
Handle the case where we don't have event log compression enabled and…
Apr 17, 2018
7da1d2d
Adjust the path to not have the .lz4 suffix if the event log files ar…
Apr 17, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 14 additions & 6 deletions app/com/linkedin/drelephant/util/SparkUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -106,10 +106,10 @@ trait SparkUtils {
appId: String,
attemptId: Option[String]
): (Path, Option[CompressionCodec]) = {
val shouldUseCompression = sparkConf.getBoolean(SPARK_EVENT_LOG_COMPRESS_KEY, defaultValue = false)
attemptId match {
// if attemptid is given, use the existing method
case x: Some[String] => { val path = {
val shouldUseCompression = sparkConf.getBoolean(SPARK_EVENT_LOG_COMPRESS_KEY, defaultValue = false)
val compressionCodecShortName =
if (shouldUseCompression) Some(shortNameOfCompressionCodec(compressionCodecFromConf(sparkConf))) else None
getLogPath(fs.getUri.resolve(basePath.toUri), appId, attemptId, compressionCodecShortName)
Expand All @@ -118,9 +118,10 @@ trait SparkUtils {
(path, codec)
}
case None => {
val (logPath, codecName) = getLogPathAndCodecName(fs, fs.getUri.resolve(basePath.toUri), appId)

(logPath, Some(compressionCodecMap.getOrElseUpdate(codecName, loadCompressionCodec(sparkConf, codecName))))
val (logPath, codecName) = getLogPathAndCodecName(fs, fs.getUri.resolve(basePath.toUri), appId, shouldUseCompression)
val codec =
if (shouldUseCompression) Some(compressionCodecMap.getOrElseUpdate(codecName, loadCompressionCodec(sparkConf, codecName))) else None
(logPath, codec)
}
}

Expand Down Expand Up @@ -241,7 +242,8 @@ trait SparkUtils {
private def getLogPathAndCodecName(
fs: FileSystem,
logBaseDir: URI,
appId: String
appId: String,
shouldUseCompression: Boolean = true
): (Path, String) = {
val base = logBaseDir.toString.stripSuffix("/");
val filter = new PathFilter() {
Expand Down Expand Up @@ -276,7 +278,13 @@ trait SparkUtils {

// This should be reached only if we can't parse the filename in the path.
// Try to construct a general path in that case.
case _ => (new Path(base + "/" + appId + "." + DEFAULT_COMPRESSION_CODEC), DEFAULT_COMPRESSION_CODEC)
case _ => {

if (shouldUseCompression)
(new Path(base + "/" + appId + "." + DEFAULT_COMPRESSION_CODEC), DEFAULT_COMPRESSION_CODEC)
else
(new Path(base + "/" + appId), DEFAULT_COMPRESSION_CODEC)
}
}
}

Expand Down