You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by da...@apache.org on 2017/09/01 19:32:32 UTC
kudu git commit: KUDU-2098. Drop Spark 1 Support
Repository: kudu
Updated Branches:
refs/heads/master ea58982e5 -> fa953d05e
KUDU-2098. Drop Spark 1 Support
Spark 2 has been available for over a year and the value of
shipping a seperate Spark 1 artifact is decreasing.
In the case where another Spark 1 release occurs or bug fixes
are needed, updates to the Kudu Spark 1 integration can still
exist via maintenance releases.
This allows us to clean up deprecated mehods that are used for
compatibility purposes and simplifies the build.
Here is the dev mailing list discussion for reference:
https://s.apache.org/PCfg
Change-Id: I5481cc0477f4d23d89b68ef510a6c9a2aa187537
Reviewed-on: http://gerrit.cloudera.org:8080/7690
Reviewed-by: Dan Burkert <da...@apache.org>
Tested-by: Kudu Jenkins
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/fa953d05
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/fa953d05
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/fa953d05
Branch: refs/heads/master
Commit: fa953d05ef933a1dc97f06a958ca30ed2eae1772
Parents: ea58982
Author: Grant Henke <gr...@gmail.com>
Authored: Wed Aug 16 14:36:47 2017 -0500
Committer: Dan Burkert <da...@apache.org>
Committed: Fri Sep 1 19:31:51 2017 +0000
----------------------------------------------------------------------
build-support/jenkins/build-and-test.sh | 8 +---
java/README.md | 29 +-----------
java/gradle/dependencies.gradle | 18 ++------
java/kudu-spark-tools/pom.xml | 8 ++--
.../kudu/spark/tools/ImportExportFiles.scala | 30 ++++++------
.../tools/IntegrationTestBigLinkedList.scala | 39 ++++++++--------
.../kudu/spark/tools/ITBigLinkedListTest.scala | 4 +-
.../spark/tools/TestImportExportFiles.scala | 9 ++--
java/kudu-spark/build.gradle | 23 +---------
java/kudu-spark/pom.xml | 18 --------
.../apache/kudu/spark/kudu/KuduContext.scala | 8 ++--
.../org/apache/kudu/spark/kudu/package.scala | 38 ++++++++++++++++
.../org/apache/kudu/spark/kudu/package.scala | 38 ----------------
.../org/apache/kudu/spark/kudu/package.scala | 38 ----------------
.../kudu/spark/kudu/DefaultSourceTest.scala | 14 +++---
.../kudu/spark/kudu/KuduContextTest.scala | 9 ++--
.../apache/kudu/spark/kudu/KuduRDDTest.scala | 2 +-
.../apache/kudu/spark/kudu/TestContext.scala | 28 ++++++------
java/pom.xml | 48 +++-----------------
19 files changed, 123 insertions(+), 286 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/fa953d05/build-support/jenkins/build-and-test.sh
----------------------------------------------------------------------
diff --git a/build-support/jenkins/build-and-test.sh b/build-support/jenkins/build-and-test.sh
index 8d828bb..b303051 100755
--- a/build-support/jenkins/build-and-test.sh
+++ b/build-support/jenkins/build-and-test.sh
@@ -375,7 +375,7 @@ if [ "$BUILD_JAVA" == "1" ]; then
export TSAN_OPTIONS="$TSAN_OPTIONS suppressions=$SOURCE_ROOT/build-support/tsan-suppressions.txt history_size=7"
set -x
- # Run the full Maven build (with Spark 2.x).
+ # Run the full Maven build.
MVN_FLAGS="$MVN_FLAGS -B"
MVN_FLAGS="$MVN_FLAGS -Dsurefire.rerunFailingTestsCount=3"
MVN_FLAGS="$MVN_FLAGS -Dfailsafe.rerunFailingTestsCount=3"
@@ -383,12 +383,6 @@ if [ "$BUILD_JAVA" == "1" ]; then
if ! mvn $MVN_FLAGS clean verify ; then
EXIT_STATUS=1
FAILURES="$FAILURES"$'Java build/test failed\n'
-
- # If there are no failures, rerun the build with Spark 1.x with Scala 2.10.
- # Note: this won't work if there are ever Spark integration tests!
- elif ! mvn $MVN_FLAGS -Dtest="org.apache.kudu.spark.*.*" -DskipITs -Pspark_2.10 clean verify ; then
- EXIT_STATUS=1
- FAILURES="$FAILURES"$'Spark 1.x build/test failed\n'
fi
# Rerun the build using the Gradle build.
http://git-wip-us.apache.org/repos/asf/kudu/blob/fa953d05/java/README.md
----------------------------------------------------------------------
diff --git a/java/README.md b/java/README.md
index 775e2b9..235b506 100644
--- a/java/README.md
+++ b/java/README.md
@@ -53,25 +53,8 @@ may be run with:
$ mvn verify
-Building the Kudu-Spark integration for Spark 2.x with Scala 2.11
-------------------------------------------------------------
-
-The Spark integration builds for Spark 2.x and Scala 2.11 by default.
-Additionally, there is a build profile available for Spark 1.x with
-Scala 2.10: from the kudu-spark directory, run
-
-$ mvn clean package -P spark_2.10
-
-The two artifactIds are
-
-1. kudu-spark2_2.11 for Spark 2.x with Scala 2.11
-2. kudu-spark_2.10 for Spark 1.x with Scala 2.10
-
-WARNING: Support for Spark 1 is deprecated as of Kudu 1.5.0 and may be removed in
-the next minor release.
-
-WARNING: Spark 2.2+ requires Java 8 at runtime even though Kudu Spark 2.x integration
-is Java 7 compatible. Spark 2.2 is the default dependency version as of Kudu 1.5.0.
+NOTE: Running the tests for the kudu-spark and kudu-spark-tools
+module requires Java 8.
State of Eclipse integration
------------------------------------------------------------
@@ -175,14 +158,6 @@ $ gradle integrationTest
*Note:* Integration tests may depend on built Kudu binaries.
-## Building the Kudu-Spark integration
-
-Builds with Spark 2.x with Scala 2.11 (the default) or
-Spark 1.x with Scala 2.10.
-
-$ gradle :kudu-spark:assemble
-$ gradle :kudu-spark:assemble -PscalaVersions=2.10.6
-
## Installing to local maven repo
$ gradle install
http://git-wip-us.apache.org/repos/asf/kudu/blob/fa953d05/java/gradle/dependencies.gradle
----------------------------------------------------------------------
diff --git a/java/gradle/dependencies.gradle b/java/gradle/dependencies.gradle
index b69f40d..153846c 100755
--- a/java/gradle/dependencies.gradle
+++ b/java/gradle/dependencies.gradle
@@ -22,8 +22,6 @@
ext {
versions = [:]
libs = [:]
- // All supported scala versions.
- scalaVersions = propertyWithDefault("scalaVersions", "2.11.11,2.10.6").split(",")
}
versions += [
@@ -48,8 +46,10 @@ versions += [
parquet : "1.9.0",
pmd : "5.8.1",
protobuf : "3.3.0",
+ scala : "2.11.11",
scalatest : "3.0.3",
slf4j : "1.7.25",
+ spark : "2.2.0",
sparkAvro : "3.2.0",
yetus : "0.5.0"
]
@@ -59,23 +59,11 @@ if (gradle.gradleVersion != versions.gradle) {
println "Using gradle version $gradle.gradleVersion (Build defines $versions.gradle)"
}
-// Set the Scala version.
-// The first Scala version in the scalaVersions list is the default.
-versions["scala"] = scalaVersions.first()
// Add base Scala version
versions["scalaBase"] = versions.scala.substring(0, versions.scala.lastIndexOf("."))
-// Set the Spark version based on the Scala version.
-def spark1Version = "1.6.3"
-def spark2Version = "2.2.0"
-if ("$versions.scalaBase" == "2.10") {
- versions["spark"] = spark1Version
-} else {
- versions["spark"] = spark2Version
-}
+// Add base Spark version
versions["sparkBase"] = versions.spark.substring(0, versions.spark.indexOf("."))
-// Log for debugging
-logger.info("Configuring Scala version: $versions.scala with Spark version: $versions.spark")
libs += [
async : "com.stumbleupon:async:$versions.async",
http://git-wip-us.apache.org/repos/asf/kudu/blob/fa953d05/java/kudu-spark-tools/pom.xml
----------------------------------------------------------------------
diff --git a/java/kudu-spark-tools/pom.xml b/java/kudu-spark-tools/pom.xml
index 10a24ca..64bfb26 100644
--- a/java/kudu-spark-tools/pom.xml
+++ b/java/kudu-spark-tools/pom.xml
@@ -44,7 +44,7 @@
</dependency>
<dependency>
<groupId>org.apache.kudu</groupId>
- <artifactId>kudu-spark2_${scala.binary.version}</artifactId>
+ <artifactId>kudu-${spark.version.label}_${scala.binary.version}</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
@@ -56,7 +56,7 @@
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_${scala.binary.version}</artifactId>
- <version>${spark2.version}</version>
+ <version>${spark.version}</version>
<scope>provided</scope>
<exclusions>
<exclusion>
@@ -74,7 +74,7 @@
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_${scala.binary.version}</artifactId>
- <version>${spark2.version}</version>
+ <version>${spark.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
@@ -93,7 +93,7 @@
</dependency>
<dependency>
<groupId>org.apache.kudu</groupId>
- <artifactId>kudu-spark2_${scala.binary.version}</artifactId>
+ <artifactId>kudu-${spark.version.label}_${scala.binary.version}</artifactId>
<version>${project.version}</version>
<type>test-jar</type>
<scope>test</scope>
http://git-wip-us.apache.org/repos/asf/kudu/blob/fa953d05/java/kudu-spark-tools/src/main/scala/org/apache/kudu/spark/tools/ImportExportFiles.scala
----------------------------------------------------------------------
diff --git a/java/kudu-spark-tools/src/main/scala/org/apache/kudu/spark/tools/ImportExportFiles.scala b/java/kudu-spark-tools/src/main/scala/org/apache/kudu/spark/tools/ImportExportFiles.scala
index bc2f0a3..025b06b 100644
--- a/java/kudu-spark-tools/src/main/scala/org/apache/kudu/spark/tools/ImportExportFiles.scala
+++ b/java/kudu-spark-tools/src/main/scala/org/apache/kudu/spark/tools/ImportExportFiles.scala
@@ -19,11 +19,10 @@ package org.apache.kudu.spark.tools
import org.apache.kudu.client.KuduClient
import org.apache.kudu.spark.tools.ImportExportKudu.ArgsCls
-import org.apache.spark.sql.SQLContext
-import org.apache.spark.{SparkConf, SparkContext}
+import org.apache.spark.sql.{SQLContext, SparkSession}
+import org.apache.spark.SparkConf
import org.slf4j.{Logger, LoggerFactory}
import org.apache.kudu.spark.kudu._
-import com.databricks.spark.avro
import com.google.common.annotations.VisibleForTesting
object ImportExportKudu {
@@ -93,21 +92,18 @@ object ImportExportKudu {
object ImportExportFiles {
- import ImportExportKudu.{LOG, fail}
+ import ImportExportKudu.fail
- var sqlContext: SQLContext = _
- var kuduOptions: Map[String, String] = _
-
- def run(args: ArgsCls, sc: SparkContext, sqlContext: SQLContext): Unit = {
- val kc = new KuduContext(args.masterAddrs, sc)
- val applicationId = sc.applicationId
+ def run(args: ArgsCls, ss: SparkSession): Unit = {
+ val kc = new KuduContext(args.masterAddrs, ss.sparkContext)
+ val sqlContext = ss.sqlContext
val client: KuduClient = kc.syncClient
if (!client.tableExists(args.tableName)) {
fail(args.tableName + s" table doesn't exist")
}
- kuduOptions = Map(
+ val kuduOptions = Map(
"kudu.table" -> args.tableName,
"kudu.master" -> args.masterAddrs)
@@ -126,7 +122,7 @@ object ImportExportFiles {
case _ => fail(args.format + s"unknown argument given ")
}
case "export" =>
- val df = sqlContext.read.options(kuduOptions).kudu.select(args.columns);
+ val df = sqlContext.read.options(kuduOptions).kudu.select(args.columns)
args.format match {
case "csv" =>
df.write.format("com.databricks.spark.csv").option("header", args.header).option("delimiter",
@@ -140,20 +136,20 @@ object ImportExportFiles {
case _ => fail(args.operation + s"unknown argument given ")
}
}
+
/**
* Entry point for testing. SparkContext is a singleton,
* so tests must create and manage their own.
*/
@VisibleForTesting
- def testMain(args: Array[String], sc: SparkContext): Unit = {
- sqlContext = new SQLContext(sc)
- run(ArgsCls.parse(args), sc, sqlContext)
+ def testMain(args: Array[String], ss: SparkSession): Unit = {
+ run(ArgsCls.parse(args), ss)
}
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setAppName("Import or Export CSV files from/to Kudu ")
- val sc = new SparkContext(conf)
- testMain(args, sc)
+ val ss = SparkSession.builder().config(conf).getOrCreate()
+ testMain(args, ss)
}
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/fa953d05/java/kudu-spark-tools/src/main/scala/org/apache/kudu/spark/tools/IntegrationTestBigLinkedList.scala
----------------------------------------------------------------------
diff --git a/java/kudu-spark-tools/src/main/scala/org/apache/kudu/spark/tools/IntegrationTestBigLinkedList.scala b/java/kudu-spark-tools/src/main/scala/org/apache/kudu/spark/tools/IntegrationTestBigLinkedList.scala
index 3c82c46..db03c24 100644
--- a/java/kudu-spark-tools/src/main/scala/org/apache/kudu/spark/tools/IntegrationTestBigLinkedList.scala
+++ b/java/kudu-spark-tools/src/main/scala/org/apache/kudu/spark/tools/IntegrationTestBigLinkedList.scala
@@ -22,8 +22,8 @@ import org.apache.kudu.client.SessionConfiguration.FlushMode
import org.apache.kudu.client.{KuduClient, KuduSession, KuduTable}
import org.apache.kudu.mapreduce.tools.BigLinkedListCommon.{Xoroshiro128PlusRandom, _}
import org.apache.kudu.spark.kudu.KuduContext
-import org.apache.spark.sql.SQLContext
-import org.apache.spark.{SparkConf, SparkContext, TaskContext}
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.{SparkConf, TaskContext}
import org.slf4j.{Logger, LoggerFactory}
import scala.util.Try
@@ -150,10 +150,9 @@ object Generator {
}
}
- def run(args: Args, sc: SparkContext): Unit = {
-
- val kc = new KuduContext(args.masterAddrs, sc)
- val applicationId = sc.applicationId
+ def run(args: Args, ss: SparkSession): Unit = {
+ val kc = new KuduContext(args.masterAddrs, ss.sparkContext)
+ val applicationId = ss.sparkContext.applicationId
val client: KuduClient = kc.syncClient
if (!client.tableExists(args.tableName)) {
@@ -164,14 +163,14 @@ object Generator {
}
// Run the generate tasks
- sc.makeRDD(0 until args.tasks, args.tasks)
+ ss.sparkContext.makeRDD(0 until args.tasks, args.tasks)
.foreach(_ => generate(args, applicationId, kc))
}
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setAppName("Integration Test Big Linked List Generator")
- val sc = new SparkContext(conf)
- run(Args.parse(args), sc)
+ val ss = SparkSession.builder().config(conf).getOrCreate()
+ run(Args.parse(args), ss)
}
/**
@@ -179,8 +178,8 @@ object Generator {
* so tests must create and manage their own.
*/
@VisibleForTesting
- def testMain(args: Array[String], sc: SparkContext): Unit = {
- run(Args.parse(args), sc)
+ def testMain(args: Array[String], ss: SparkSession): Unit = {
+ run(Args.parse(args), ss)
}
def generate(args: Args,
@@ -332,9 +331,9 @@ object Verifier {
}
@VisibleForTesting
- def run(args: Args, sc: SparkContext): Counts = {
+ def run(args: Args, ss: SparkSession): Counts = {
import org.apache.kudu.spark.kudu._
- val sql = new SQLContext(sc)
+ val sql = ss.sqlContext
sql.read
.option("kudu.master", args.masterAddrs)
@@ -414,16 +413,16 @@ object Verifier {
}
@VisibleForTesting
- def testMain(arguments: Array[String], sc: SparkContext): Counts = {
- run(Args.parse(arguments), sc)
+ def testMain(arguments: Array[String], ss: SparkSession): Counts = {
+ run(Args.parse(arguments), ss)
}
def main(arguments: Array[String]): Unit = {
val args = Args.parse(arguments)
val conf = new SparkConf().setAppName("Integration Test Big Linked List Generator")
- val sc = new SparkContext(conf)
+ val ss = SparkSession.builder().config(conf).getOrCreate()
- val counts = run(Args.parse(arguments), sc)
+ val counts = run(Args.parse(arguments), ss)
verify(args.nodes, counts).map(fail)
}
}
@@ -432,7 +431,7 @@ object Looper {
import IntegrationTestBigLinkedList.{LOG, fail}
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setAppName("Integration Test Big Linked List Looper")
- val sc = new SparkContext(conf)
+ val ss = SparkSession.builder().config(conf).getOrCreate()
val genArgs = Generator.Args.parse(args)
var verifyArgs = Verifier.Args(masterAddrs = genArgs.masterAddrs,
@@ -440,8 +439,8 @@ object Looper {
val nodesPerLoop = genArgs.tasks * genArgs.lists * genArgs.nodes
for (n <- Stream.from(1)) {
- Generator.run(genArgs, sc)
- val count = Verifier.run(verifyArgs, sc)
+ Generator.run(genArgs, ss)
+ val count = Verifier.run(verifyArgs, ss)
val expected = verifyArgs.nodes.map(_ + nodesPerLoop)
Verifier.verify(expected, count).map(fail)
verifyArgs = verifyArgs.copy(nodes = Some(expected.getOrElse(nodesPerLoop)))
http://git-wip-us.apache.org/repos/asf/kudu/blob/fa953d05/java/kudu-spark-tools/src/test/scala/org/apache/kudu/spark/tools/ITBigLinkedListTest.scala
----------------------------------------------------------------------
diff --git a/java/kudu-spark-tools/src/test/scala/org/apache/kudu/spark/tools/ITBigLinkedListTest.scala b/java/kudu-spark-tools/src/test/scala/org/apache/kudu/spark/tools/ITBigLinkedListTest.scala
index 6b34a5a..8e1d7ef 100644
--- a/java/kudu-spark-tools/src/test/scala/org/apache/kudu/spark/tools/ITBigLinkedListTest.scala
+++ b/java/kudu-spark-tools/src/test/scala/org/apache/kudu/spark/tools/ITBigLinkedListTest.scala
@@ -38,7 +38,7 @@ class ITBigLinkedListTest extends FunSuite with TestContext with Matchers {
"--range-partitions=2",
"--replicas=1",
s"--master-addrs=${miniCluster.getMasterAddresses}"),
- sc)
+ ss)
// Insert bad nodes in order to test the verifier:
//
@@ -75,7 +75,7 @@ class ITBigLinkedListTest extends FunSuite with TestContext with Matchers {
}
}
- val counts = Verifier.testMain(Array(s"--master-addrs=${miniCluster.getMasterAddresses}"), sc)
+ val counts = Verifier.testMain(Array(s"--master-addrs=${miniCluster.getMasterAddresses}"), ss)
assertEquals(2 * 2 * 10000, counts.referenced)
assertEquals(1, counts.extrareferences)
assertEquals(2, counts.unreferenced)
http://git-wip-us.apache.org/repos/asf/kudu/blob/fa953d05/java/kudu-spark-tools/src/test/scala/org/apache/kudu/spark/tools/TestImportExportFiles.scala
----------------------------------------------------------------------
diff --git a/java/kudu-spark-tools/src/test/scala/org/apache/kudu/spark/tools/TestImportExportFiles.scala b/java/kudu-spark-tools/src/test/scala/org/apache/kudu/spark/tools/TestImportExportFiles.scala
index 2507853..6031b6a 100644
--- a/java/kudu-spark-tools/src/test/scala/org/apache/kudu/spark/tools/TestImportExportFiles.scala
+++ b/java/kudu-spark-tools/src/test/scala/org/apache/kudu/spark/tools/TestImportExportFiles.scala
@@ -23,7 +23,6 @@ import org.apache.kudu.ColumnSchema.ColumnSchemaBuilder
import org.apache.kudu.{Schema, Type}
import org.apache.kudu.client.CreateTableOptions
import org.apache.kudu.spark.kudu._
-import org.apache.spark.sql.SQLContext
import org.junit.Assert._
import org.junit.runner.RunWith
import org.scalatest.junit.JUnitRunner
@@ -36,8 +35,6 @@ import scala.collection.JavaConverters._
class TestImportExportFiles extends FunSuite with TestContext with Matchers {
private val TABLE_NAME: String = classOf[TestImportExportFiles].getName + "-" + System.currentTimeMillis
- var sqlContext : SQLContext = _
- var kuduOptions : Map[String, String] = _
test("Spark Import Export") {
val schema: Schema = {
@@ -60,11 +57,11 @@ class TestImportExportFiles extends FunSuite with TestContext with Matchers {
"--format=csv",
s"--master-addrs=${miniCluster.getMasterAddresses}",
s"--path=${"target/"+TABLE_NAME+".csv"}",
- s"--table-name=${TABLE_NAME}",
+ s"--table-name=$TABLE_NAME",
"--delimiter=,",
"--header=true",
- "--inferschema=true"), sc)
- val rdd = kuduContext.kuduRDD(sc, TABLE_NAME, List("key"))
+ "--inferschema=true"), ss)
+ val rdd = kuduContext.kuduRDD(ss.sparkContext, TABLE_NAME, List("key"))
assert(rdd.collect.length == 4)
assertEquals(rdd.collect().mkString(","),"[1],[2],[3],[4]")
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/fa953d05/java/kudu-spark/build.gradle
----------------------------------------------------------------------
diff --git a/java/kudu-spark/build.gradle b/java/kudu-spark/build.gradle
index 0ce65ac..0c89a8b 100644
--- a/java/kudu-spark/build.gradle
+++ b/java/kudu-spark/build.gradle
@@ -31,24 +31,5 @@ dependencies {
testCompile libs.scalatest
}
-// Add compatibility sources based on the spark version used.
-sourceSets {
- main {
- scala {
- if (versions.sparkBase == "1") {
- srcDir "src/main/spark1"
- } else if (versions.sparkBase == "2") {
- srcDir "src/main/spark2"
- } else {
- throw new GradleException("No compatibility sources exist for spark base version ${versions.sparkBase}")
- }
- }
- }
-}
-
-// Adjust the artifact name to match the maven build.
-if (versions.sparkBase == "1") {
- archivesBaseName = "kudu-spark_${versions.scalaBase}"
-} else {
- archivesBaseName = "kudu-spark${versions.sparkBase}_${versions.scalaBase}"
-}
\ No newline at end of file
+// Adjust the artifact name to include the spark and scala base versions.
+archivesBaseName = "kudu-spark${versions.sparkBase}_${versions.scalaBase}"
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/kudu/blob/fa953d05/java/kudu-spark/pom.xml
----------------------------------------------------------------------
diff --git a/java/kudu-spark/pom.xml b/java/kudu-spark/pom.xml
index 6974616..c579b5a 100644
--- a/java/kudu-spark/pom.xml
+++ b/java/kudu-spark/pom.xml
@@ -144,24 +144,6 @@
</artifactSet>
</configuration>
</plugin>
-
- <!-- http://stackoverflow.com/questions/270445 -->
- <plugin>
- <groupId>org.codehaus.mojo</groupId>
- <artifactId>build-helper-maven-plugin</artifactId>
- <version>${build-helper-maven-plugin.version}</version>
- <executions>
- <execution>
- <phase>generate-sources</phase>
- <goals><goal>add-source</goal></goals>
- <configuration>
- <sources>
- <source>${kudu-spark.compat.src}</source>
- </sources>
- </configuration>
- </execution>
- </executions>
- </plugin>
</plugins>
</build>
</project>
http://git-wip-us.apache.org/repos/asf/kudu/blob/fa953d05/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/KuduContext.scala
----------------------------------------------------------------------
diff --git a/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/KuduContext.scala b/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/KuduContext.scala
index ad6d0bd..71eb6e8 100644
--- a/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/KuduContext.scala
+++ b/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/KuduContext.scala
@@ -55,7 +55,7 @@ class KuduContext(val kuduMaster: String,
this(kuduMaster, new SparkContext())
}
- @transient lazy val syncClient = {
+ @transient lazy val syncClient: KuduClient = {
val c = KuduConnection.getSyncClient(kuduMaster)
if (authnCredentials != null) {
c.importAuthenticationCredentials(authnCredentials)
@@ -63,7 +63,7 @@ class KuduContext(val kuduMaster: String,
c
}
- @transient lazy val asyncClient = {
+ @transient lazy val asyncClient: AsyncKuduClient = {
val c = KuduConnection.getAsyncClient(kuduMaster)
if (authnCredentials != null) {
c.importAuthenticationCredentials(authnCredentials)
@@ -325,7 +325,7 @@ private object KuduConnection {
if (!syncCache.contains(kuduMaster)) {
val syncClient = new KuduClient.KuduClientBuilder(kuduMaster).build()
ShutdownHookManager.get().addShutdownHook(new Runnable {
- override def run() = syncClient.close()
+ override def run(): Unit = syncClient.close()
}, ShutdownHookPriority)
syncCache.put(kuduMaster, syncClient)
}
@@ -339,7 +339,7 @@ private object KuduConnection {
val asyncClient = new AsyncKuduClient.AsyncKuduClientBuilder(kuduMaster).build()
ShutdownHookManager.get().addShutdownHook(
new Runnable {
- override def run() = asyncClient.close()
+ override def run(): Unit = asyncClient.close()
}, ShutdownHookPriority)
asyncCache.put(kuduMaster, asyncClient)
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/fa953d05/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/package.scala
----------------------------------------------------------------------
diff --git a/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/package.scala b/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/package.scala
new file mode 100755
index 0000000..19ae9ef
--- /dev/null
+++ b/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/package.scala
@@ -0,0 +1,38 @@
+/*
+ * 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 org.apache.kudu.spark
+
+import org.apache.spark.sql.{DataFrame, DataFrameReader, DataFrameWriter}
+
+package object kudu {
+
+ /**
+ * Adds a method, `kudu`, to DataFrameReader that allows you to read Kudu tables using
+ * the DataFrameReader.
+ */
+ implicit class KuduDataFrameReader(reader: DataFrameReader) {
+ def kudu: DataFrame = reader.format("org.apache.kudu.spark.kudu").load
+ }
+
+ /**
+ * Adds a method, `kudu`, to DataFrameWriter that allows writes to Kudu using
+ * the DataFileWriter
+ */
+ implicit class KuduDataFrameWriter[T](writer: DataFrameWriter[T]) {
+ def kudu = writer.format("org.apache.kudu.spark.kudu").save
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/kudu/blob/fa953d05/java/kudu-spark/src/main/spark1/org/apache/kudu/spark/kudu/package.scala
----------------------------------------------------------------------
diff --git a/java/kudu-spark/src/main/spark1/org/apache/kudu/spark/kudu/package.scala b/java/kudu-spark/src/main/spark1/org/apache/kudu/spark/kudu/package.scala
deleted file mode 100755
index 6f49023..0000000
--- a/java/kudu-spark/src/main/spark1/org/apache/kudu/spark/kudu/package.scala
+++ /dev/null
@@ -1,38 +0,0 @@
-/*
- * 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 org.apache.kudu.spark
-
-import org.apache.spark.sql.{DataFrame, DataFrameReader, DataFrameWriter}
-
-package object kudu {
-
- /**
- * Adds a method, `kudu`, to DataFrameReader that allows you to read Kudu tables using
- * the DataFrameReader.
- */
- implicit class KuduDataFrameReader(reader: DataFrameReader) {
- def kudu: DataFrame = reader.format("org.apache.kudu.spark.kudu").load
- }
-
- /**
- * Adds a method, `kudu`, to DataFrameWriter that allows writes to Kudu using
- * the DataFileWriter
- */
- implicit class KuduDataFrameWriter(writer: DataFrameWriter) {
- def kudu = writer.format("org.apache.kudu.spark.kudu").save
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/kudu/blob/fa953d05/java/kudu-spark/src/main/spark2/org/apache/kudu/spark/kudu/package.scala
----------------------------------------------------------------------
diff --git a/java/kudu-spark/src/main/spark2/org/apache/kudu/spark/kudu/package.scala b/java/kudu-spark/src/main/spark2/org/apache/kudu/spark/kudu/package.scala
deleted file mode 100755
index 19ae9ef..0000000
--- a/java/kudu-spark/src/main/spark2/org/apache/kudu/spark/kudu/package.scala
+++ /dev/null
@@ -1,38 +0,0 @@
-/*
- * 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 org.apache.kudu.spark
-
-import org.apache.spark.sql.{DataFrame, DataFrameReader, DataFrameWriter}
-
-package object kudu {
-
- /**
- * Adds a method, `kudu`, to DataFrameReader that allows you to read Kudu tables using
- * the DataFrameReader.
- */
- implicit class KuduDataFrameReader(reader: DataFrameReader) {
- def kudu: DataFrame = reader.format("org.apache.kudu.spark.kudu").load
- }
-
- /**
- * Adds a method, `kudu`, to DataFrameWriter that allows writes to Kudu using
- * the DataFileWriter
- */
- implicit class KuduDataFrameWriter[T](writer: DataFrameWriter[T]) {
- def kudu = writer.format("org.apache.kudu.spark.kudu").save
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/kudu/blob/fa953d05/java/kudu-spark/src/test/scala/org/apache/kudu/spark/kudu/DefaultSourceTest.scala
----------------------------------------------------------------------
diff --git a/java/kudu-spark/src/test/scala/org/apache/kudu/spark/kudu/DefaultSourceTest.scala b/java/kudu-spark/src/test/scala/org/apache/kudu/spark/kudu/DefaultSourceTest.scala
index ea9d916..a65c530 100644
--- a/java/kudu-spark/src/test/scala/org/apache/kudu/spark/kudu/DefaultSourceTest.scala
+++ b/java/kudu-spark/src/test/scala/org/apache/kudu/spark/kudu/DefaultSourceTest.scala
@@ -67,13 +67,13 @@ class DefaultSourceTest extends FunSuite with TestContext with BeforeAndAfter wi
before {
rows = insertRows(rowCount)
- sqlContext = new SQLContext(sc)
+ sqlContext = ss.sqlContext
kuduOptions = Map(
"kudu.table" -> tableName,
"kudu.master" -> miniCluster.getMasterAddresses)
- sqlContext.read.options(kuduOptions).kudu.registerTempTable(tableName)
+ sqlContext.read.options(kuduOptions).kudu.createOrReplaceTempView(tableName)
}
test("table creation") {
@@ -213,7 +213,7 @@ class DefaultSourceTest extends FunSuite with TestContext with BeforeAndAfter wi
"kudu.faultTolerantScan" -> "true")
val table = "faultTolerantScanTest"
- sqlContext.read.options(kuduOptions).kudu.registerTempTable(table)
+ sqlContext.read.options(kuduOptions).kudu.createOrReplaceTempView(table)
val results = sqlContext.sql(s"SELECT * FROM $table").collectAsList()
assert(results.size() == rowCount)
@@ -358,7 +358,7 @@ class DefaultSourceTest extends FunSuite with TestContext with BeforeAndAfter wi
val testTable = kuduClient.createTable(testTableName, schema, tableOptions)
val kuduSession = kuduClient.newSession()
- val chars = List('a', 'b', '乕', Char.MaxValue, '\0')
+ val chars = List('a', 'b', '乕', Char.MaxValue, '\u0000')
val keys = for (x <- chars; y <- chars; z <- chars; w <- chars) yield Array(x, y, z, w).mkString
keys.foreach { key =>
val insert = testTable.newInsert
@@ -370,7 +370,7 @@ class DefaultSourceTest extends FunSuite with TestContext with BeforeAndAfter wi
val options: Map[String, String] = Map(
"kudu.table" -> testTableName,
"kudu.master" -> miniCluster.getMasterAddresses)
- sqlContext.read.options(options).kudu.registerTempTable(testTableName)
+ sqlContext.read.options(options).kudu.createOrReplaceTempView(testTableName)
val checkPrefixCount = { prefix: String =>
val results = sqlContext.sql(s"SELECT key FROM $testTableName WHERE key LIKE '$prefix%'")
@@ -416,7 +416,7 @@ class DefaultSourceTest extends FunSuite with TestContext with BeforeAndAfter wi
val newOptions: Map[String, String] = Map(
"kudu.table" -> insertTable,
"kudu.master" -> miniCluster.getMasterAddresses)
- sqlContext.read.options(newOptions).kudu.registerTempTable(insertTable)
+ sqlContext.read.options(newOptions).kudu.createOrReplaceTempView(insertTable)
sqlContext.sql(s"INSERT INTO TABLE $insertTable SELECT * FROM $tableName")
val results = sqlContext.sql(s"SELECT key FROM $insertTable").collectAsList()
@@ -435,7 +435,7 @@ class DefaultSourceTest extends FunSuite with TestContext with BeforeAndAfter wi
val newOptions: Map[String, String] = Map(
"kudu.table" -> insertTable,
"kudu.master" -> miniCluster.getMasterAddresses)
- sqlContext.read.options(newOptions).kudu.registerTempTable(insertTable)
+ sqlContext.read.options(newOptions).kudu.createOrReplaceTempView(insertTable)
try {
sqlContext.sql(s"INSERT OVERWRITE TABLE $insertTable SELECT * FROM $tableName")
http://git-wip-us.apache.org/repos/asf/kudu/blob/fa953d05/java/kudu-spark/src/test/scala/org/apache/kudu/spark/kudu/KuduContextTest.scala
----------------------------------------------------------------------
diff --git a/java/kudu-spark/src/test/scala/org/apache/kudu/spark/kudu/KuduContextTest.scala b/java/kudu-spark/src/test/scala/org/apache/kudu/spark/kudu/KuduContextTest.scala
index 6ec218b..620cf3f 100644
--- a/java/kudu-spark/src/test/scala/org/apache/kudu/spark/kudu/KuduContextTest.scala
+++ b/java/kudu-spark/src/test/scala/org/apache/kudu/spark/kudu/KuduContextTest.scala
@@ -19,7 +19,6 @@ package org.apache.kudu.spark.kudu
import java.io.{ByteArrayInputStream, ByteArrayOutputStream, ObjectInputStream, ObjectOutputStream}
import java.sql.Timestamp
-import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.functions.decode
import org.junit.runner.RunWith
import org.scalatest.junit.JUnitRunner
@@ -36,7 +35,7 @@ class KuduContextTest extends FunSuite with TestContext with Matchers {
oos.writeObject(value)
stream.toByteArray
} finally {
- oos.close
+ oos.close()
}
}
@@ -45,7 +44,7 @@ class KuduContextTest extends FunSuite with TestContext with Matchers {
try {
ois.readObject
} finally {
- ois.close
+ ois.close()
}
}
@@ -61,7 +60,7 @@ class KuduContextTest extends FunSuite with TestContext with Matchers {
test("Test basic kuduRDD") {
val rows = insertRows(rowCount)
- val scanList = kuduContext.kuduRDD(sc, "test", Seq("key", "c1_i", "c2_s", "c3_double",
+ val scanList = kuduContext.kuduRDD(ss.sparkContext, "test", Seq("key", "c1_i", "c2_s", "c3_double",
"c4_long", "c5_bool", "c6_short", "c7_float", "c8_binary", "c9_unixtime_micros", "c10_byte"))
.map(r => r.toSeq).collect()
scanList.foreach(r => {
@@ -84,7 +83,7 @@ class KuduContextTest extends FunSuite with TestContext with Matchers {
test("Test kudu-spark DataFrame") {
insertRows(rowCount)
- val sqlContext = new SQLContext(sc)
+ val sqlContext = ss.sqlContext
val dataDF = sqlContext.read.options(Map("kudu.master" -> miniCluster.getMasterAddresses,
"kudu.table" -> "test")).kudu
dataDF.sort("key").select("c8_binary").first.get(0)
http://git-wip-us.apache.org/repos/asf/kudu/blob/fa953d05/java/kudu-spark/src/test/scala/org/apache/kudu/spark/kudu/KuduRDDTest.scala
----------------------------------------------------------------------
diff --git a/java/kudu-spark/src/test/scala/org/apache/kudu/spark/kudu/KuduRDDTest.scala b/java/kudu-spark/src/test/scala/org/apache/kudu/spark/kudu/KuduRDDTest.scala
index 1ecac34..fbd7bbb 100644
--- a/java/kudu-spark/src/test/scala/org/apache/kudu/spark/kudu/KuduRDDTest.scala
+++ b/java/kudu-spark/src/test/scala/org/apache/kudu/spark/kudu/KuduRDDTest.scala
@@ -26,7 +26,7 @@ class KuduRDDTest extends FunSuite with TestContext with BeforeAndAfter {
test("collect rows") {
insertRows(100)
- val rdd = kuduContext.kuduRDD(sc, tableName, List("key"))
+ val rdd = kuduContext.kuduRDD(ss.sparkContext, tableName, List("key"))
assert(rdd.collect.length == 100)
}
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/fa953d05/java/kudu-spark/src/test/scala/org/apache/kudu/spark/kudu/TestContext.scala
----------------------------------------------------------------------
diff --git a/java/kudu-spark/src/test/scala/org/apache/kudu/spark/kudu/TestContext.scala b/java/kudu-spark/src/test/scala/org/apache/kudu/spark/kudu/TestContext.scala
index 2748f9a..6a0c08d 100644
--- a/java/kudu-spark/src/test/scala/org/apache/kudu/spark/kudu/TestContext.scala
+++ b/java/kudu-spark/src/test/scala/org/apache/kudu/spark/kudu/TestContext.scala
@@ -22,7 +22,7 @@ import scala.collection.JavaConverters._
import scala.collection.immutable.IndexedSeq
import com.google.common.collect.ImmutableList
-import org.apache.spark.{SparkConf, SparkContext}
+import org.apache.spark.SparkConf
import org.scalatest.{BeforeAndAfterAll, Suite}
import org.apache.kudu.ColumnSchema.ColumnSchemaBuilder
@@ -30,16 +30,17 @@ import org.apache.kudu.client.KuduClient.KuduClientBuilder
import org.apache.kudu.client.MiniKuduCluster.MiniKuduClusterBuilder
import org.apache.kudu.client.{CreateTableOptions, KuduClient, KuduTable, MiniKuduCluster}
import org.apache.kudu.{Schema, Type}
+import org.apache.spark.sql.SparkSession
trait TestContext extends BeforeAndAfterAll { self: Suite =>
- var sc: SparkContext = null
- var miniCluster: MiniKuduCluster = null
- var kuduClient: KuduClient = null
- var table: KuduTable = null
- var kuduContext: KuduContext = null
+ var ss: SparkSession = _
+ var miniCluster: MiniKuduCluster = _
+ var kuduClient: KuduClient = _
+ var table: KuduTable = _
+ var kuduContext: KuduContext = _
- val tableName = "test"
+ val tableName: String = "test"
lazy val schema: Schema = {
val columns = ImmutableList.of(
@@ -57,9 +58,9 @@ trait TestContext extends BeforeAndAfterAll { self: Suite =>
new Schema(columns)
}
- val appID = new Date().toString + math.floor(math.random * 10E4).toLong.toString
+ val appID: String = new Date().toString + math.floor(math.random * 10E4).toLong.toString
- val conf = new SparkConf().
+ val conf: SparkConf = new SparkConf().
setMaster("local[*]").
setAppName("test").
set("spark.ui.enabled", "false").
@@ -70,14 +71,13 @@ trait TestContext extends BeforeAndAfterAll { self: Suite =>
.numMasters(1)
.numTservers(1)
.build()
- val envMap = Map[String,String](("Xmx", "512m"))
- sc = new SparkContext(conf)
+ ss = SparkSession.builder().config(conf).getOrCreate()
kuduClient = new KuduClientBuilder(miniCluster.getMasterAddresses).build()
assert(miniCluster.waitForTabletServers(1))
- kuduContext = new KuduContext(miniCluster.getMasterAddresses, sc)
+ kuduContext = new KuduContext(miniCluster.getMasterAddresses, ss.sparkContext)
val tableOptions = new CreateTableOptions().setRangePartitionColumns(List("key").asJava)
.setNumReplicas(1)
@@ -87,7 +87,7 @@ trait TestContext extends BeforeAndAfterAll { self: Suite =>
override def afterAll() {
if (kuduClient != null) kuduClient.shutdown()
if (miniCluster != null) miniCluster.shutdown()
- if (sc != null) sc.stop()
+ if (ss != null) ss.stop()
}
def deleteRow(key: Int): Unit = {
@@ -110,7 +110,7 @@ trait TestContext extends BeforeAndAfterAll { self: Suite =>
row.addBoolean(5, i%2==1)
row.addShort(6, i.toShort)
row.addFloat(7, i.toFloat)
- row.addBinary(8, s"bytes ${i}".getBytes())
+ row.addBinary(8, s"bytes $i".getBytes())
val ts = System.currentTimeMillis() * 1000
row.addLong(9, ts)
row.addByte(10, i.toByte)
http://git-wip-us.apache.org/repos/asf/kudu/blob/fa953d05/java/pom.xml
----------------------------------------------------------------------
diff --git a/java/pom.xml b/java/pom.xml
index dd7636d..c01f97a 100644
--- a/java/pom.xml
+++ b/java/pom.xml
@@ -48,7 +48,6 @@
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
<!-- Maven plugins -->
- <build-helper-maven-plugin.version>3.0.0</build-helper-maven-plugin.version>
<clojure-maven-plugin.version>1.7.1</clojure-maven-plugin.version>
<maven-antrun-plugin.version>1.8</maven-antrun-plugin.version>
<maven-assembly-plugin.version>3.0.0</maven-assembly-plugin.version>
@@ -89,11 +88,11 @@
<yetus.version>0.4.0</yetus.version>
<!-- Scala Library dependencies -->
- <scala-2.10.version>2.10.6</scala-2.10.version>
- <scala-2.11.version>2.11.11</scala-2.11.version>
+ <scala.binary.version>2.11</scala.binary.version>
+ <scala.version>2.11.11</scala.version>
<scalatest.version>3.0.3</scalatest.version>
- <spark1.version>1.6.3</spark1.version>
- <spark2.version>2.2.0</spark2.version>
+ <spark.version>2.1.1</spark.version>
+ <spark.version.label>spark2</spark.version.label>
<!-- Misc variables -->
<testdata.dir>target/testdata</testdata.dir>
@@ -108,6 +107,8 @@
<module>kudu-client-tools</module>
<module>kudu-mapreduce</module>
<module>kudu-flume-sink</module>
+ <module>kudu-spark</module>
+ <module>kudu-spark-tools</module>
</modules>
<build>
@@ -336,42 +337,5 @@
<module>kudu-jepsen</module>
</modules>
</profile>
-
- <!-- Manage Scala and Spark versions.
-
- Kudu supports integration with Spark2/Scala2.11 *and* Spark1/Scala2.10
- with the same module, so we use profiles to control which mode to compile in. -->
- <profile>
- <id>spark2_2.11</id>
- <activation>
- <activeByDefault>true</activeByDefault>
- </activation>
- <properties>
- <spark.version>${spark2.version}</spark.version>
- <spark.version.label>spark2</spark.version.label>
- <scala.version>${scala-2.11.version}</scala.version>
- <scala.binary.version>2.11</scala.binary.version>
- <kudu-spark.compat.src>src/main/spark2</kudu-spark.compat.src>
- </properties>
- <modules>
- <module>kudu-spark</module>
-
- <!-- kudu-spark-tools is only compatible with Spark2 -->
- <module>kudu-spark-tools</module>
- </modules>
- </profile>
- <profile>
- <id>spark_2.10</id>
- <properties>
- <spark.version>${spark1.version}</spark.version>
- <spark.version.label>spark</spark.version.label>
- <scala.version>${scala-2.10.version}</scala.version>
- <scala.binary.version>2.10</scala.binary.version>
- <kudu-spark.compat.src>src/main/spark1</kudu-spark.compat.src>
- </properties>
- <modules>
- <module>kudu-spark</module>
- </modules>
- </profile>
</profiles>
</project>