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 2016/02/02 21:15:15 UTC
incubator-kudu git commit: Spark connectors for Kudu
Repository: incubator-kudu
Updated Branches:
refs/heads/master 266d214b1 -> ba3a17a2f
Spark connectors for Kudu
This is a cleanup of Ted Malaska's "SparkOnKudu":
https://github.com/tmalaska/SparkOnKudu
Even if this patch is very simple and unfinished, the goal is to
provide a platform for others to add more functionality.
It contains wrappers around common Spark operators that make it easy
to use Kudu. It also has a Data Source implementation.
Basic scala unit tests are provided.
Change-Id: Ic187513ef9724d50024f7401d7ecd19d53554245
Reviewed-on: http://gerrit.cloudera.org:8080/1788
Tested-by: Kudu Jenkins
Reviewed-by: Jean-Daniel Cryans
Project: http://git-wip-us.apache.org/repos/asf/incubator-kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-kudu/commit/ba3a17a2
Tree: http://git-wip-us.apache.org/repos/asf/incubator-kudu/tree/ba3a17a2
Diff: http://git-wip-us.apache.org/repos/asf/incubator-kudu/diff/ba3a17a2
Branch: refs/heads/master
Commit: ba3a17a2f12ae546d290fd03e75b5e0c347787dc
Parents: 266d214
Author: Jean-Daniel Cryans <jd...@cloudera.com>
Authored: Thu Jan 14 16:10:59 2016 -0800
Committer: Dan Burkert <da...@cloudera.com>
Committed: Tue Feb 2 18:11:33 2016 +0000
----------------------------------------------------------------------
java/kudu-spark/pom.xml | 208 +++++++++++++++++++
.../scala/org/kududb/spark/DefaultSource.scala | 148 +++++++++++++
.../scala/org/kududb/spark/KuduContext.scala | 67 ++++++
.../src/test/resources/log4j.properties | 23 ++
.../org/kududb/spark/DefaultSourceSuite.scala | 37 ++++
.../org/kududb/spark/KuduContextSuite.scala | 32 +++
.../scala/org/kududb/spark/TestContext.scala | 82 ++++++++
java/pom.xml | 1 +
8 files changed, 598 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/ba3a17a2/java/kudu-spark/pom.xml
----------------------------------------------------------------------
diff --git a/java/kudu-spark/pom.xml b/java/kudu-spark/pom.xml
new file mode 100644
index 0000000..51a1f0c
--- /dev/null
+++ b/java/kudu-spark/pom.xml
@@ -0,0 +1,208 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+// Licensed 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.kududb</groupId>
+ <artifactId>kudu-parent</artifactId>
+ <version>0.7.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>kudu-spark</artifactId>
+ <name>Kudu Spark Bindings</name>
+
+ <properties>
+ <spark.version>1.3.0</spark.version>
+ <scala.version>2.10.4</scala.version>
+ <scala.binary.version>2.10</scala.binary.version>
+ <top.dir>${project.basedir}/..</top.dir>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-core_${scala.binary.version}</artifactId>
+ <version>${spark.version}</version>
+ <scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <!-- make sure wrong scala version is not pulled in -->
+ <groupId>org.scala-lang</groupId>
+ <artifactId>scala-library</artifactId>
+ </exclusion>
+ <exclusion>
+ <!-- make sure wrong scala version is not pulled in -->
+ <groupId>org.scala-lang</groupId>
+ <artifactId>scalap</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-sql_${scala.binary.version}</artifactId>
+ <version>${spark.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-streaming_${scala.binary.version}</artifactId>
+ <version>${spark.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-streaming_${scala.binary.version}</artifactId>
+ <version>${spark.version}</version>
+ <type>test-jar</type>
+ <classifier>tests</classifier>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.kududb</groupId>
+ <artifactId>kudu-client</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.kududb</groupId>
+ <artifactId>kudu-client</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.kududb</groupId>
+ <artifactId>interface-annotations</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.kududb</groupId>
+ <artifactId>kudu-mapreduce</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ <version>${hadoop.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.scala-lang</groupId>
+ <artifactId>scala-library</artifactId>
+ <version>${scala.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <version>${junit.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.scalatest</groupId>
+ <artifactId>scalatest_${scala.binary.version}</artifactId>
+ <version>2.2.4</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.scalamock</groupId>
+ <artifactId>scalamock-scalatest-support_${scala.binary.version}</artifactId>
+ <version>3.1.4</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ </plugin>
+
+ <plugin>
+ <groupId>net.alchim31.maven</groupId>
+ <artifactId>scala-maven-plugin</artifactId>
+ <version>3.2.0</version>
+ <configuration>
+ <charset>${project.build.sourceEncoding}</charset>
+ <scalaVersion>${scala.version}</scalaVersion>
+ </configuration>
+ <executions>
+ <execution>
+ <id>scala-compile-first</id>
+ <phase>process-resources</phase>
+ <goals>
+ <goal>add-source</goal>
+ <goal>compile</goal>
+ </goals>
+ </execution>
+ <execution>
+ <id>scala-test-compile</id>
+ <phase>process-test-resources</phase>
+ <goals>
+ <goal>testCompile</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+
+ <plugin>
+ <!-- disable surefire -->
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <skipTests>true</skipTests>
+ </configuration>
+ </plugin>
+
+ <plugin>
+ <groupId>org.scalatest</groupId>
+ <artifactId>scalatest-maven-plugin</artifactId>
+ <version>1.0</version>
+ <configuration>
+ <reportsDirectory>${project.build.directory}/surefire-reports</reportsDirectory>
+ <junitxml>.</junitxml>
+ <filereports>WDF TestSuite.txt</filereports>
+ <parallel>false</parallel>
+ </configuration>
+ <executions>
+ <execution>
+ <id>test</id>
+ <phase>test</phase>
+ <goals>
+ <goal>test</goal>
+ </goals>
+ <configuration>
+ <skipTests>false</skipTests>
+ </configuration>
+ </execution>
+ <execution>
+ <id>integration-test</id>
+ <phase>integration-test</phase>
+ <goals>
+ <goal>test</goal>
+ </goals>
+ <configuration>
+ <tagsToExclude>Integration-Test</tagsToExclude>
+ <parallel>false</parallel>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/ba3a17a2/java/kudu-spark/src/main/scala/org/kududb/spark/DefaultSource.scala
----------------------------------------------------------------------
diff --git a/java/kudu-spark/src/main/scala/org/kududb/spark/DefaultSource.scala b/java/kudu-spark/src/main/scala/org/kududb/spark/DefaultSource.scala
new file mode 100644
index 0000000..50cb9c6
--- /dev/null
+++ b/java/kudu-spark/src/main/scala/org/kududb/spark/DefaultSource.scala
@@ -0,0 +1,148 @@
+/*
+ * 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.kududb.spark
+
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.sources._
+import org.apache.spark.sql.types._
+import org.apache.spark.sql.{Row, SQLContext}
+import org.kududb.Type
+import org.kududb.annotations.InterfaceStability
+import org.kududb.client.RowResult
+
+import scala.collection.JavaConverters._
+import scala.collection.immutable.HashMap
+
+/**
+ * DefaultSource for integration with Spark's dataframe datasources.
+ * This class will produce a relationProvider based on input given to it from spark.
+ */
+@InterfaceStability.Unstable
+class DefaultSource extends RelationProvider {
+
+ val TABLE_KEY = "kudu.table"
+ val KUDU_MASTER = "kudu.master"
+
+ /**
+ * Construct a BaseRelation using the provided context and parameters.
+ *
+ * @param sqlContext SparkSQL context
+ * @param parameters parameters given to us from SparkSQL
+ * @return a BaseRelation Object
+ */
+ override def createRelation(sqlContext: SQLContext,
+ parameters: Map[String, String]):
+ BaseRelation = {
+ val tableName = parameters.get(TABLE_KEY)
+ if (tableName.isEmpty) {
+ throw new IllegalArgumentException(s"Invalid value for $TABLE_KEY '$tableName'")
+ }
+
+ val kuduMaster = parameters.getOrElse(KUDU_MASTER, "localhost")
+
+ new KuduRelation(tableName.get, kuduMaster)(sqlContext)
+ }
+}
+
+/**
+ * Implementation of Spark BaseRelation.
+ *
+ * @param tableName Kudu table that we plan to read from
+ * @param kuduMaster Kudu master addresses
+ * @param sqlContext SparkSQL context
+ */
+@InterfaceStability.Unstable
+class KuduRelation(val tableName: String,
+ val kuduMaster: String)(
+ @transient val sqlContext: SQLContext)
+ extends BaseRelation with PrunedFilteredScan with Serializable {
+
+ val typesMapping = HashMap[Type, DataType](
+ Type.INT16 -> IntegerType,
+ Type.INT32 -> IntegerType,
+ Type.INT64 -> LongType,
+ Type.FLOAT -> FloatType,
+ Type.DOUBLE -> DoubleType,
+ Type.STRING -> StringType,
+ Type.TIMESTAMP -> TimestampType,
+ Type.BINARY -> BinaryType
+ )
+
+ // Using lazy val for the following because we can't serialize them but we need them once we
+ // deserialize them.
+ @transient lazy val kuduContext = new KuduContext(kuduMaster)
+ @transient lazy val kuduTable = kuduContext.syncClient.openTable(tableName)
+ @transient lazy val tableColumns = kuduTable.getSchema.getColumns.asScala
+ @transient lazy val kuduSchemaColumnMap = tableColumns.map(c => (c.getName, c)).toMap
+
+ /**
+ * Generates a SparkSQL schema object so SparkSQL knows what is being
+ * provided by this BaseRelation.
+ *
+ * @return schema generated from the Kudu table's schema
+ */
+ override def schema: StructType = {
+ val metadataBuilder = new MetadataBuilder()
+
+ val structFieldArray: Array[StructField] =
+ tableColumns.map { columnSchema =>
+ val columnSparkSqlType = typesMapping.getOrElse(
+ columnSchema.getType,
+ throw new IllegalArgumentException(s"Unsupported column type: ${columnSchema.getType}"))
+
+ val metadata = metadataBuilder.putString("name", columnSchema.getName).build()
+ new StructField(columnSchema.getName, columnSparkSqlType,
+ nullable = columnSchema.isNullable, metadata)
+ }.toArray
+
+ new StructType(structFieldArray)
+ }
+
+ /**
+ * Build the RDD to scan rows.
+ *
+ * @param requiredColumns clumns that are being requested by the requesting query
+ * @param filters filters that are being applied by the requesting query
+ * @return RDD will all the results from Kudu
+ */
+ override def buildScan(requiredColumns: Array[String], filters: Array[Filter]): RDD[Row] = {
+ kuduContext.kuduRDD(sqlContext.sparkContext, tableName, requiredColumns).map { row =>
+ // TODO use indexes instead of column names since it requires one less mapping.
+ Row.fromSeq(requiredColumns.map(column => getKuduValue(row, column)))
+ }
+ }
+
+ private def getKuduValue(row: RowResult, columnName: String): Any = {
+ val columnType = kuduSchemaColumnMap.getOrElse(columnName,
+ throw new IllegalArgumentException(s"Couldn't find column '$columnName'")).getType
+
+ columnType match {
+ case Type.BINARY => row.getBinary(columnName)
+ case Type.BOOL => row.getBoolean(columnName)
+ case Type.DOUBLE => row.getDouble(columnName)
+ case Type.FLOAT => row.getFloat(columnName)
+ case Type.INT16 => row.getShort(columnName)
+ case Type.INT32 => row.getInt(columnName)
+ case Type.INT64 => row.getLong(columnName)
+ case Type.INT8 => row.getByte(columnName)
+ case Type.TIMESTAMP => row.getLong(columnName)
+ case Type.STRING => row.getString(columnName)
+ case _ => throw new IllegalArgumentException(s"Type not supported: '${columnType.getName}'")
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/ba3a17a2/java/kudu-spark/src/main/scala/org/kududb/spark/KuduContext.scala
----------------------------------------------------------------------
diff --git a/java/kudu-spark/src/main/scala/org/kududb/spark/KuduContext.scala b/java/kudu-spark/src/main/scala/org/kududb/spark/KuduContext.scala
new file mode 100644
index 0000000..a034099
--- /dev/null
+++ b/java/kudu-spark/src/main/scala/org/kududb/spark/KuduContext.scala
@@ -0,0 +1,67 @@
+/*
+ * 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.kududb.spark
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.io.NullWritable
+import org.apache.spark.rdd.RDD
+import org.apache.spark.SparkContext
+import org.kududb.annotations.InterfaceStability
+import org.kududb.client.{AsyncKuduClient, KuduClient, RowResult}
+import org.kududb.mapreduce.KuduTableInputFormat
+
+/**
+ * KuduContext is a façade for Kudu operations.
+ *
+ * If a Kudu client connection is needed as part of a Spark application, a
+ * [[KuduContext]] should used as a broadcast variable in the job in order to
+ * share connections among the tasks in a JVM.
+ */
+@InterfaceStability.Unstable
+class KuduContext(kuduMaster: String) extends Serializable {
+ @transient lazy val syncClient = new KuduClient.KuduClientBuilder(kuduMaster).build()
+ @transient lazy val asyncClient = new AsyncKuduClient.AsyncKuduClientBuilder(kuduMaster).build()
+
+ /**
+ * Create an RDD from a Kudu table.
+ *
+ * @param tableName table to read from
+ * @param columnProjection list of columns to read
+ *
+ * Not specifying this at all (i.e. setting to null) or setting to the special string
+ * '*' means to project all columns.
+ * @return a new RDD that maps over the given table for the selected columns
+ */
+ def kuduRDD(sc: SparkContext,
+ tableName: String,
+ columnProjection: Seq[String] = Nil): RDD[RowResult] = {
+
+ val conf = new Configuration
+ conf.set("kudu.mapreduce.master.address", kuduMaster)
+ conf.set("kudu.mapreduce.input.table", tableName)
+ if (columnProjection.nonEmpty) {
+ conf.set("kudu.mapreduce.column.projection", columnProjection.mkString(","))
+ }
+
+ val rdd = sc.newAPIHadoopRDD(conf, classOf[KuduTableInputFormat],
+ classOf[NullWritable], classOf[RowResult])
+
+ val columnNames = if (columnProjection.nonEmpty) columnProjection.mkString(", ") else "(*)"
+ rdd.values.setName(s"KuduRDD { table=$tableName, columnProjection=$columnNames }")
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/ba3a17a2/java/kudu-spark/src/test/resources/log4j.properties
----------------------------------------------------------------------
diff --git a/java/kudu-spark/src/test/resources/log4j.properties b/java/kudu-spark/src/test/resources/log4j.properties
new file mode 100644
index 0000000..cb277ed
--- /dev/null
+++ b/java/kudu-spark/src/test/resources/log4j.properties
@@ -0,0 +1,23 @@
+# 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.
+
+log4j.rootLogger = DEBUG, out
+log4j.appender.out = org.apache.log4j.ConsoleAppender
+log4j.appender.out.layout = org.apache.log4j.PatternLayout
+log4j.appender.out.layout.ConversionPattern = %d (%t) [%p - %l] %m%n
+
+log4j.logger.org.kududb = DEBUG
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/ba3a17a2/java/kudu-spark/src/test/scala/org/kududb/spark/DefaultSourceSuite.scala
----------------------------------------------------------------------
diff --git a/java/kudu-spark/src/test/scala/org/kududb/spark/DefaultSourceSuite.scala b/java/kudu-spark/src/test/scala/org/kududb/spark/DefaultSourceSuite.scala
new file mode 100644
index 0000000..820caec
--- /dev/null
+++ b/java/kudu-spark/src/test/scala/org/kududb/spark/DefaultSourceSuite.scala
@@ -0,0 +1,37 @@
+/*
+ * 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.kududb.spark
+
+import org.apache.spark.sql.SQLContext
+import org.scalatest.FunSuite
+
+class DefaultSourceSuite extends FunSuite with TestContext {
+
+ test("Test basic SparkSQL") {
+ val rowCount = 10
+
+ insertRows(rowCount)
+
+ val sqlContext = new SQLContext(sc)
+
+ sqlContext.load("org.kududb.spark",
+ Map("kudu.table" -> tableName, "kudu.master" -> miniCluster.getMasterAddresses))
+ .registerTempTable(tableName)
+
+ assert(sqlContext.sql("SELECT * FROM " + tableName).collectAsList().size() == rowCount)
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/ba3a17a2/java/kudu-spark/src/test/scala/org/kududb/spark/KuduContextSuite.scala
----------------------------------------------------------------------
diff --git a/java/kudu-spark/src/test/scala/org/kududb/spark/KuduContextSuite.scala b/java/kudu-spark/src/test/scala/org/kududb/spark/KuduContextSuite.scala
new file mode 100644
index 0000000..67f8c89
--- /dev/null
+++ b/java/kudu-spark/src/test/scala/org/kududb/spark/KuduContextSuite.scala
@@ -0,0 +1,32 @@
+/*
+ * 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.kududb.spark
+
+import org.scalatest.FunSuite
+
+class KuduContextSuite extends FunSuite with TestContext {
+ test("Test basic kuduRDD") {
+ val rowCount = 10
+
+ insertRows(rowCount)
+
+ val scanRdd = kuduContext.kuduRDD(sc, "test")
+
+ val scanList = scanRdd.map(r => r.getInt(0)).collect()
+ assert(scanList.length == rowCount)
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/ba3a17a2/java/kudu-spark/src/test/scala/org/kududb/spark/TestContext.scala
----------------------------------------------------------------------
diff --git a/java/kudu-spark/src/test/scala/org/kududb/spark/TestContext.scala b/java/kudu-spark/src/test/scala/org/kududb/spark/TestContext.scala
new file mode 100644
index 0000000..fa24f52
--- /dev/null
+++ b/java/kudu-spark/src/test/scala/org/kududb/spark/TestContext.scala
@@ -0,0 +1,82 @@
+/*
+ * 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.kududb.spark
+
+import com.google.common.collect.ImmutableList
+import org.apache.spark.SparkContext
+import org.kududb.ColumnSchema.ColumnSchemaBuilder
+import org.kududb.client.KuduClient.KuduClientBuilder
+import org.kududb.client.MiniKuduCluster.MiniKuduClusterBuilder
+import org.kududb.client.{CreateTableOptions, KuduClient, KuduTable, MiniKuduCluster}
+import org.kududb.{Schema, Type}
+import org.scalatest.{BeforeAndAfterAll, Suite}
+
+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
+
+ val tableName = "test"
+
+ lazy val schema: Schema = {
+ val columns = ImmutableList.of(
+ new ColumnSchemaBuilder("key", Type.INT32).key(true).build(),
+ new ColumnSchemaBuilder("c1_i", Type.INT32).build(),
+ new ColumnSchemaBuilder("c2_s", Type.STRING).build())
+ new Schema(columns)
+ }
+
+ override def beforeAll() {
+ miniCluster = new MiniKuduClusterBuilder()
+ .numMasters(1)
+ .numTservers(1)
+ .build()
+ val envMap = Map[String,String](("Xmx", "512m"))
+
+ sc = new SparkContext("local[2]", "test", null, Nil, envMap)
+
+ kuduClient = new KuduClientBuilder(miniCluster.getMasterAddresses).build()
+ assert(miniCluster.waitForTabletServers(1))
+
+ kuduContext = new KuduContext(miniCluster.getMasterAddresses)
+
+ val tableOptions = new CreateTableOptions().setNumReplicas(1)
+ table = kuduClient.createTable(tableName, schema, tableOptions)
+ }
+
+ override def afterAll() {
+ if (kuduClient != null) kuduClient.shutdown()
+ if (miniCluster != null) miniCluster.shutdown()
+ if (sc != null) sc.stop()
+ }
+
+ def insertRows(rowCount: Integer) {
+ val kuduSession = kuduClient.newSession()
+
+ for (i <- 1 to rowCount) {
+ val insert = table.newInsert
+ val row = insert.getRow
+ row.addInt(0, i)
+ row.addInt(1, i)
+ row.addString(2, i.toString)
+ kuduSession.apply(insert)
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/ba3a17a2/java/pom.xml
----------------------------------------------------------------------
diff --git a/java/pom.xml b/java/pom.xml
index f441fe7..52df900 100644
--- a/java/pom.xml
+++ b/java/pom.xml
@@ -73,6 +73,7 @@
<module>kudu-client</module>
<module>kudu-client-tools</module>
<module>kudu-mapreduce</module>
+ <module>kudu-spark</module>
</modules>
<build>