[SPARK-33084][CORE][SQL] Add jar support ivy path #29966
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -15,7 +15,7 @@ | |
| * limitations under the License. | ||
| */ | ||
|
|
||
| package org.apache.spark.util | ||
|
|
||
| import java.io.File | ||
| import java.net.URI | ||
|
|
@@ -25,12 +25,140 @@ import org.apache.hadoop.conf.Configuration | |
| import org.apache.hadoop.fs.{FileSystem, Path} | ||
|
|
||
| import org.apache.spark.{SecurityManager, SparkConf, SparkException} | ||
| import org.apache.spark.deploy.SparkSubmitUtils | ||
| import org.apache.spark.internal.Logging | ||
|
|
||
| case class IvyProperties( | ||
| packagesExclusions: String, | ||
| packages: String, | ||
| repositories: String, | ||
| ivyRepoPath: String, | ||
| ivySettingsPath: String) | ||
|
|
||
| private[spark] object DependencyUtils extends Logging { | ||
|
|
||
| def getIvyProperties(): IvyProperties = { | ||
| val Seq(packagesExclusions, packages, repositories, ivyRepoPath, ivySettingsPath) = Seq( | ||
| "spark.jars.excludes", | ||
| "spark.jars.packages", | ||
| "spark.jars.repositories", | ||
| "spark.jars.ivy", | ||
| "spark.jars.ivySettings" | ||
| ).map(sys.props.get(_).orNull) | ||
| IvyProperties(packagesExclusions, packages, repositories, ivyRepoPath, ivySettingsPath) | ||
| } | ||
|
|
||
AngersZhuuuu marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| private def isInvalidQueryString(tokens: Array[String]): Boolean = { | ||
| tokens.length != 2 || StringUtils.isBlank(tokens(0)) || StringUtils.isBlank(tokens(1)) | ||
| } | ||
|
|
||
| /** | ||
| * Parse URI query string's parameter value of `transitive` and `exclude`. | ||
| * Other invalid parameters will be ignored. | ||
| * | ||
| * @param uri Ivy URI need to be downloaded. | ||
| * @return Tuple value of parameter `transitive` and `exclude` value. | ||
| * | ||
| * 1. transitive: whether to download dependency jar of Ivy URI, default value is false | ||
| * and this parameter value is case-sensitive. Invalid value will be treat as false. | ||
| * Example: Input: exclude=org.mortbay.jetty:jetty&transitive=true | ||
| * Output: true | ||
| * | ||
| * 2. exclude: comma separated exclusions to apply when resolving transitive dependencies, | ||
| * consists of `group:module` pairs separated by commas. | ||
| * Example: Input: excludeorg.mortbay.jetty:jetty,org.eclipse.jetty:jetty-http | ||
| * Output: [org.mortbay.jetty:jetty,org.eclipse.jetty:jetty-http] | ||
| */ | ||
| private def parseQueryParams(uri: URI): (Boolean, String) = { | ||
| val uriQuery = uri.getQuery | ||
| if (uriQuery == null) { | ||
| (false, "") | ||
| } else { | ||
| val mapTokens = uriQuery.split("&").map(_.split("=")) | ||
| if (mapTokens.exists(isInvalidQueryString)) { | ||
| throw new IllegalArgumentException( | ||
| s"Invalid query string in Ivy URI ${uri.toString}: $uriQuery") | ||
| } | ||
| val groupedParams = mapTokens.map(kv => (kv(0), kv(1))).groupBy(_._1) | ||
|
|
||
| // Parse transitive parameters (e.g., transitive=true) in an Ivy URI, default value is false | ||
| val transitiveParams = groupedParams.get("transitive") | ||
| if (transitiveParams.map(_.size).getOrElse(0) > 1) { | ||
| logWarning("It's best to specify `transitive` parameter in ivy URI query only once." + | ||
| " If there are multiple `transitive` parameter, we will select the last one") | ||
|
There was a problem hiding this comment. hive has the same behaviour with this? There was a problem hiding this comment.
No, but we can have this warning There was a problem hiding this comment. If so, what's the hive behaviour? it will throw an exception instead selecting the last one? There was a problem hiding this comment.
Select the last one |
||
| } | ||
| val transitive = | ||
| transitiveParams.flatMap(_.takeRight(1).map(_._2 == "true").headOption).getOrElse(false) | ||
AngersZhuuuu marked this conversation as resolved.
Show resolved
Hide resolved
|
||
|
|
||
| // Parse an excluded list (e.g., exclude=org.mortbay.jetty:jetty,org.eclipse.jetty:jetty-http) | ||
| // in an Ivy URI. When download Ivy URI jar, Spark won't download transitive jar | ||
| // in a excluded list. | ||
| val exclusionList = groupedParams.get("exclude").map { params => | ||
| params.map(_._2).flatMap { excludeString => | ||
| val excludes = excludeString.split(",") | ||
| if (excludes.map(_.split(":")).exists(isInvalidQueryString)) { | ||
| throw new IllegalArgumentException( | ||
| s"Invalid exclude string in Ivy URI ${uri.toString}:" + | ||
| " expected 'org:module,org:module,..', found " + excludeString) | ||
| } | ||
| excludes | ||
| }.mkString(",") | ||
| }.getOrElse("") | ||
|
|
||
| val validParams = Set("transitive", "exclude") | ||
| val invalidParams = groupedParams.keys.filterNot(validParams.contains).toSeq | ||
| if (invalidParams.nonEmpty) { | ||
| logWarning(s"Invalid parameters `${invalidParams.sorted.mkString(",")}` found " + | ||
| s"in Ivy URI query `$uriQuery`.") | ||
| } | ||
|
|
||
| (transitive, exclusionList) | ||
|
There was a problem hiding this comment. What if an invalid param is given in hive, e.g., There was a problem hiding this comment. Seems like we should at least warn on invalid param? |
||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Download Ivy URI's dependency jars. | ||
| * | ||
| * @param uri Ivy URI need to be downloaded. The URI format should be: | ||
| * `ivy://group:module:version[?query]` | ||
| * Ivy URI query part format should be: | ||
| * `parameter=value¶meter=value...` | ||
| * Note that currently Ivy URI query part support two parameters: | ||
| * 1. transitive: whether to download dependent jars related to your Ivy URI. | ||
| * transitive=false or `transitive=true`, if not set, the default value is false. | ||
| * 2. exclude: exclusion list when download Ivy URI jar and dependency jars. | ||
| * The `exclude` parameter content is a ',' separated `group:module` pair string : | ||
| * `exclude=group:module,group:module...` | ||
| * @return Comma separated string list of jars downloaded. | ||
| */ | ||
| def resolveMavenDependencies(uri: URI): Seq[String] = { | ||
| val ivyProperties = DependencyUtils.getIvyProperties() | ||
| val authority = uri.getAuthority | ||
| if (authority == null) { | ||
| throw new IllegalArgumentException( | ||
| s"Invalid Ivy URI authority in uri ${uri.toString}:" + | ||
| " Expected 'org:module:version', found null.") | ||
| } | ||
| if (authority.split(":").length != 3) { | ||
| throw new IllegalArgumentException( | ||
| s"Invalid Ivy URI authority in uri ${uri.toString}:" + | ||
| s" Expected 'org:module:version', found $authority.") | ||
| } | ||
|
|
||
| val (transitive, exclusionList) = parseQueryParams(uri) | ||
|
|
||
| resolveMavenDependencies( | ||
| transitive, | ||
| exclusionList, | ||
| authority, | ||
| ivyProperties.repositories, | ||
| ivyProperties.ivyRepoPath, | ||
| Option(ivyProperties.ivySettingsPath) | ||
| ).split(",") | ||
| } | ||
|
|
||
| def resolveMavenDependencies( | ||
| packagesTransitive: Boolean, | ||
| packagesExclusions: String, | ||
| packages: String, | ||
| repositories: String, | ||
|
|
@@ -51,7 +179,8 @@ private[deploy] object DependencyUtils extends Logging { | |
| SparkSubmitUtils.buildIvySettings(Option(repositories), Option(ivyRepoPath)) | ||
| } | ||
|
|
||
| SparkSubmitUtils.resolveMavenCoordinates(packages, ivySettings, | ||
| transitive = packagesTransitive, exclusions = exclusions) | ||
| } | ||
|
|
||
| def resolveAndDownloadJars( | ||
|
|
||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@AngersZhuuuu, out of curiosity, is the Ivy URI the standard form documented somewhere? or something specific to Spark that you came up with?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
From hive https://issues.apache.org/jira/browse/HIVE-9664, since it download jar use
ivythen use schema asivy? I think this useful for a lot of companies that have standard package management, so I implemented it in Spark