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>