Merge pull request #590 from waitinfuture/support-complex-type

support spark3

Author: powerwu

README.md

@@ -23,6 +23,13 @@
 
 ```
 
+## Build emr-maxcompute with Spark3.2.0
+```
+ git clone https://github.com/aliyun/aliyun-emapreduce-datasources.git
+ cd aliyun-emapreduce-datasources/emr-maxcompute/
+ mvn clean package -Pspark3 -DskipTests
+```
+
 #### Use SDK in Eclipse project directly
 
 - copy sdk jar to your project

emr-maxcompute/pom.xml

@@ -46,6 +46,13 @@
  <type>test-jar</type>
  </dependency>
 
+ <dependency>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-lang3</artifactId>
+ <version>${commons.lang3.version}</version>
+ <scope>test</scope>
+ </dependency>
+
  <dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-sql_${scala.binary.version}</artifactId>

emr-maxcompute/src/main/scala/org/apache/spark/aliyun/odps/OdpsRDD.scala

@@ -17,20 +17,17 @@
 package org.apache.spark.aliyun.odps
 
 import java.io.EOFException
-
 import scala.collection.mutable.ArrayBuffer
 import scala.reflect.ClassTag
-
 import com.aliyun.odps.{Odps, PartitionSpec, TableSchema}
 import com.aliyun.odps.account.AliyunAccount
 import com.aliyun.odps.data.Record
 import com.aliyun.odps.tunnel.TableTunnel
 import com.aliyun.odps.tunnel.io.TunnelRecordReader
-
 import org.apache.spark._
 import org.apache.spark.internal.Logging
 import org.apache.spark.rdd.RDD
-import org.apache.spark.util.NextIterator
+import org.apache.spark.util.{NextIterator, TaskCompletionListener}
 
 class OdpsPartition(rddId: Int,
  idx: Int,
@@ -90,9 +87,11 @@ class OdpsRDD[T: ClassTag](@transient sc: SparkContext,
  val reader = downloadSession.openRecordReader(split.start, split.count)
  val inputMetrics = context.taskMetrics.inputMetrics
 
- context.addTaskCompletionListener {
- context => closeIfNeeded()
- }
+ context.addTaskCompletionListener(new TaskCompletionListener {
+ override def onTaskCompletion(context: TaskContext): Unit = {
+ closeIfNeeded()
+ }
+ })
 
  override def getNext() = {
  var ret = null.asInstanceOf[T]

emr-maxcompute/src/main/scala/org/apache/spark/aliyun/odps/datasource/ODPSRDD.scala

@@ -29,7 +29,7 @@ import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions.SpecificInternalRow
 import org.apache.spark.sql.types._
-import org.apache.spark.util.NextIterator
+import org.apache.spark.util.{NextIterator, TaskCompletionListener}
 
 class ODPSRDD(
  sc: SparkContext,
@@ -66,9 +66,11 @@ class ODPSRDD(
  val reader = downloadSession.openRecordReader(split.start, split.count)
  val inputMetrics = context.taskMetrics.inputMetrics
 
- context.addTaskCompletionListener {
- context => closeIfNeeded()
- }
+ context.addTaskCompletionListener(new TaskCompletionListener {
+ override def onTaskCompletion(context: TaskContext): Unit = {
+ closeIfNeeded()
+ }
+ })
 
  val mutableRow = new SpecificInternalRow(schema.fields.map(x => x.dataType))
 

emr-maxcompute/src/main/scala/org/apache/spark/aliyun/odps/datasource/ODPSWriter.scala

@@ -151,16 +151,15 @@ class ODPSWriter(
  writer.close()
  }
 
- data.foreachPartition {
- iterator =>
+ data.foreachPartition((iterator: Iterator[Row]) => {
  val account_ = new AliyunAccount(accessKeyId, accessKeySecret)
  val odps = new Odps(account_)
  odps.setDefaultProject(project)
  odps.setEndpoint(odpsUrl)
  val odpsUtils = new OdpsUtils(odps)
  val dataSchema = odpsUtils.getTableSchema(project, table, false)
  writeToFile(odps, dataSchema, iterator)
- }
+ })
  val arr = Array.tabulate(data.rdd.partitions.length)(l => Long.box(l))
  uploadSession.commit(arr)
  }

emr-maxcompute/src/test/scala/org/apache/spark/sql/aliyun/odps/OdpsOpsSuite.scala

@@ -94,7 +94,7 @@ class OdpsOpsSuite extends SparkFunSuite {
  odpsUtils.runSQL(project, "TRUNCATE TABLE odps_basic_types;")
  }
 
- test("support basic types") {
+ test("[OdpsOpsSuite] support basic types") {
  val table = "odps_basic_types"
  val struct = StructType(
  StructField("a", BooleanType, true) ::

emr-maxcompute/src/test/scala/org/apache/spark/sql/aliyun/odps/types/OdpsDataTypeSuite.scala

@@ -90,7 +90,7 @@ class OdpsDataTypeSuite extends SparkFunSuite {
  odpsUtils.runSQL(project, "TRUNCATE TABLE odps_basic_types;")
  }
 
- test("support basic types") {
+ test("[OdpsDataTypeSuite] support basic types") {
  val table = "odps_basic_types"
  val struct = StructType(
  StructField("a", BooleanType, true) ::

pom.xml

@@ -795,6 +795,17 @@
  <activeByDefault>true</activeByDefault>
  </activation>
  </profile>
+ <profile>
+ <id>spark3</id>
+ <properties>
+ <scala.version>2.12.15</scala.version>
+ <scala.binary.version>2.12</scala.binary.version>
+ <scalatest.version>3.2.9</scalatest.version>
+ <spark.version>3.2.0</spark.version>
+ <hadoop.version>3.3.1</hadoop.version>
+ <commons.lang3.version>3.12.0</commons.lang3.version>
+ </properties>
+ </profile>
 
  <profile>
  <id>ds</id>