Merge pull request #19 from sgrG24/master

Upgrading Scala and Spark version.

Author: ashwanthkumar

.travis.yml

@@ -1,34 +1,34 @@
 language: scala
 scala:
-- 2.10.4
-- 2.11.11
-jdk: oraclejdk8
+  - 2.11.11
+  - 2.12.11
+jdk: openjdk8
 sudo: false
 cache:
   directories:
-  - "$HOME/.ivy2/cache"
-  - "$HOME/.sbt"
+    - "$HOME/.ivy2/cache"
+    - "$HOME/.sbt"
 before_cache:
-- find $HOME/.ivy2/cache -name "ivydata-*.properties" -print -delete
-- find $HOME/.sbt        -name "*.lock"               -print -delete
+  - find $HOME/.ivy2/cache -name "ivydata-*.properties" -print -delete
+  - find $HOME/.sbt        -name "*.lock"               -print -delete
 env:
   global:
-  - BUILD_LABEL=1.0.${TRAVIS_BUILD_NUMBER}
-  - secure: pQPYXlJp86IIELAHf48ri2Q7oDISCgA137fT9wTiozlu+5YjwY9wUa+OEbDvKG1RBeqK5aWFP8KWRfoD1CEpTeLg8h+ElvX+JRhUuQS8GaX1Ev4i79YOxnLP1d/K74SEpqM3dWpacuZzP2TlYv9nGSuxC7U2PpxP54ShjBPeaAw3G7lES/vFPIdVhDXjNEjwpjfl/Fy9AanBf1f7souX/7GXVmdvg+NIDkp0ALryD7hfz38HYEd6B1RrLqqNgVFqrbSoh9bmPA9BXt8gjytEAaKkb3hB7X5Sl0Yw3xwaZzNEkxwZSSP/Z80vUJZpwAnr3TAc54yiVKx64N0CoH/kzv7FGX2fn5P4tFmwEYnsqZZvJXQ64fILHOzqiTKho2qSXA1DRsd6RD/DUME24PEpmZPmFCj7ujNuc6L15oFqIU4mSiog6HJFn8dXZAWHWJNgfgipbe/7jT8PwCigMtFsrSPgiKau39rZ3wL9jmhzUGqsGEWaWapuXdcyrG+Erzngh7Z5VI/c6Z7OLQ3zIpDBFXYenB6PV+gFqvH8Cafmb08h85a3Jnp4Va8ml9GJlTf1aNgaITSxzCEQO1RZZgrwS5Z+fu4N9J8hFWk/mou3r9nOazJ+wkGCcudQaBhl70gHluM09yaFRF5+7NjwcTMvRmdGkfA1tCnwsF/xVghci2I=
-  - secure: akbuJ5hN9AWCL+pdsgxV6dUtg318SWK3TgUg9mJ/3ix7L6aNQWb670bYJHA9VCGxX4glzRdzJttu+7PVebbngPpdXTu6TAE3CabZbV0rarPqszvtRPBQmXtA/n36TO4VwHBJdCT3UaOxp2j3pocjqxHujZ1WPS5oUg/LY07HN6y/dgwC01j7CYK6Se7//QUs45KuKcyZkTCZO4qj4Mb+fyUehWoMaq0Ivlflj91CPr/QsIBw/y2zIKlQD/Z07EQx0fVwdhkcapH3iOj2dNoKO40Jx8B2/64OpHnH3w8ycA+QdTWgmrhGbJnhyA+WXVSPTyzIluSJFiseTJPuNH1fJ5sBlN27fPngiEEy9sFhRo/ia1riZwOB1iLqWWv1pquBJHHXYy/PwTWMee8trIjw6LVgTp4tBeLLqoHXJv5z6qubakRbqc8sxSVLXEV+DyxQJ0bcn3ywCTLikhABuRIJPfwK+ekrS3b5KLnw/9mbyaSa/UP2Ko7KAQtdjKeJz5MZ8yRTdO1PI804wUqJKjLQyka9FXY3FvbO8vjkGO1uK7nny19NRDwB65nMD/HywhA2NwHNnyzlpFtw3ZCIJ/qUNW0Qnp34YYBAAYHbq/nTtOklBcPDB3qYW5WA59nPqeU+1RzMZOnYHNKEl5ahtdlituKSy/RrwTh9zMH5eQsGiYA=
-  - secure: FR3Aq3XQ4mWH8ie4Srcjf8oph7VBZF6JZ0ujHNb7DzPCWWc7X/tEf2Ae8DSpmb55PEui7/EMULmeaFXNbyIfVZ9weyCFBGEw7xZ2Ue6pF9T+bHz4KgNIPkbNAgxTrfxHqdTXcackDR/qSU40+hEOWzrgHYAegIWm7iGIIHeFnuKc3Ee7OK1iF5jN5ujwU4fK4m4N/gstdHiFAJ0Or0gmy2+i+GauwjkDB73zOIakjyV/TYwI4CJqqXEbXtc2rT4uOXGpPhgc69OmJplr9oIEn6FKzqLJmM+kmZPmuMp0cT2I8ga6Q4xtAeWj47Z+h0LbU27uqK4LkwDR5ucukRaGDB750WG3c+216m9xJCom63yn2hmUOmkiWU6lKqR8DeE0DqwiGjN/Q6Vr1z+oQ5EeSxAT4+J2xpDOhDkceyx5DpZgXOMyk7y0Y+78KdQVUouVczJFqCubMzenGR7rNzn9CHMkBpg07+14MXKqz7XQIPUB74Q8qwu44JH6rl5kGnf9bgcoJi7xVI25wse0wmvBdkGw7GQQsjq9WCX6pzf6zz40g9nnNGbAGCtgNaANX3pvGp+V5cczv2hwW9fP0/RnZd+lCyA+zT+n2qY8jCEDBogMi4+cMNHVIQ35hHkdlhHdwdFjsTJoKsuXCS6YZZD1B/GQmCPBJcHH3X+HH9pdGpU=
-  - secure: TTmBqJ0U2DrADusK6GYCGQKytnCJUqeqS+2XuE+TRp3pECdpx/arSrGit5j3oBzr2yAiakk/6N6JtWEZU7ftYaOxG7ga1Chs9Bno5T+uL67uFWkqrwxvg4AYgTUXsO+Xel7V1YmiNrXctQeIILEMB7H8OaLc6/cis3laMxY/drwexJjmf249boqfHKJuv+kHwaqfjikGO3uCLZqQsDLgzRSVH/yNvPmgUu/DZXmF6uHcfGqQRra7/VfhzFvXqAhTriYhfUyzKgouWerlcmgscN8uS9SSF8j2W+iXvR1DXyQbbf9/OV7kp+NLVTPg1wvL/RyOiQC8cMTFkdcD0lQ5xl6BBKorFPltk2Zh0Qt7syd7MJ6AMRBjoGckya6EPfTjW3rh0Pr/UDuqrVVqrP0xjgp10LUB72oW5QSDNZPXNdXgj+x9ugYtLZJIes7+yYF7ALlm7R/0RlNNiIlA6HHdOrM3Fx8BY9MHWTtqYGNxZ3yZsYZVdCk2aOyGS5o9DVXLFWWuxDFwsjokA/zc5PstIsBbyJvBqCrfoZ8YQ6b31KcdXKsN+vF+3jJP4p8nDvvz9zeIVYWGYSYrNCpRfZtLWyIyXOSDpfJ+iEgTMlxgyWuZd6dIQ3TdgcqwQhYnF5z/Nl7GgpraIvT4ttaaJkInSm6rgEA9PZr573nIWY10p28=
+    - BUILD_LABEL=1.0.${TRAVIS_BUILD_NUMBER}
+    - secure: pQPYXlJp86IIELAHf48ri2Q7oDISCgA137fT9wTiozlu+5YjwY9wUa+OEbDvKG1RBeqK5aWFP8KWRfoD1CEpTeLg8h+ElvX+JRhUuQS8GaX1Ev4i79YOxnLP1d/K74SEpqM3dWpacuZzP2TlYv9nGSuxC7U2PpxP54ShjBPeaAw3G7lES/vFPIdVhDXjNEjwpjfl/Fy9AanBf1f7souX/7GXVmdvg+NIDkp0ALryD7hfz38HYEd6B1RrLqqNgVFqrbSoh9bmPA9BXt8gjytEAaKkb3hB7X5Sl0Yw3xwaZzNEkxwZSSP/Z80vUJZpwAnr3TAc54yiVKx64N0CoH/kzv7FGX2fn5P4tFmwEYnsqZZvJXQ64fILHOzqiTKho2qSXA1DRsd6RD/DUME24PEpmZPmFCj7ujNuc6L15oFqIU4mSiog6HJFn8dXZAWHWJNgfgipbe/7jT8PwCigMtFsrSPgiKau39rZ3wL9jmhzUGqsGEWaWapuXdcyrG+Erzngh7Z5VI/c6Z7OLQ3zIpDBFXYenB6PV+gFqvH8Cafmb08h85a3Jnp4Va8ml9GJlTf1aNgaITSxzCEQO1RZZgrwS5Z+fu4N9J8hFWk/mou3r9nOazJ+wkGCcudQaBhl70gHluM09yaFRF5+7NjwcTMvRmdGkfA1tCnwsF/xVghci2I=
+    - secure: akbuJ5hN9AWCL+pdsgxV6dUtg318SWK3TgUg9mJ/3ix7L6aNQWb670bYJHA9VCGxX4glzRdzJttu+7PVebbngPpdXTu6TAE3CabZbV0rarPqszvtRPBQmXtA/n36TO4VwHBJdCT3UaOxp2j3pocjqxHujZ1WPS5oUg/LY07HN6y/dgwC01j7CYK6Se7//QUs45KuKcyZkTCZO4qj4Mb+fyUehWoMaq0Ivlflj91CPr/QsIBw/y2zIKlQD/Z07EQx0fVwdhkcapH3iOj2dNoKO40Jx8B2/64OpHnH3w8ycA+QdTWgmrhGbJnhyA+WXVSPTyzIluSJFiseTJPuNH1fJ5sBlN27fPngiEEy9sFhRo/ia1riZwOB1iLqWWv1pquBJHHXYy/PwTWMee8trIjw6LVgTp4tBeLLqoHXJv5z6qubakRbqc8sxSVLXEV+DyxQJ0bcn3ywCTLikhABuRIJPfwK+ekrS3b5KLnw/9mbyaSa/UP2Ko7KAQtdjKeJz5MZ8yRTdO1PI804wUqJKjLQyka9FXY3FvbO8vjkGO1uK7nny19NRDwB65nMD/HywhA2NwHNnyzlpFtw3ZCIJ/qUNW0Qnp34YYBAAYHbq/nTtOklBcPDB3qYW5WA59nPqeU+1RzMZOnYHNKEl5ahtdlituKSy/RrwTh9zMH5eQsGiYA=
+    - secure: FR3Aq3XQ4mWH8ie4Srcjf8oph7VBZF6JZ0ujHNb7DzPCWWc7X/tEf2Ae8DSpmb55PEui7/EMULmeaFXNbyIfVZ9weyCFBGEw7xZ2Ue6pF9T+bHz4KgNIPkbNAgxTrfxHqdTXcackDR/qSU40+hEOWzrgHYAegIWm7iGIIHeFnuKc3Ee7OK1iF5jN5ujwU4fK4m4N/gstdHiFAJ0Or0gmy2+i+GauwjkDB73zOIakjyV/TYwI4CJqqXEbXtc2rT4uOXGpPhgc69OmJplr9oIEn6FKzqLJmM+kmZPmuMp0cT2I8ga6Q4xtAeWj47Z+h0LbU27uqK4LkwDR5ucukRaGDB750WG3c+216m9xJCom63yn2hmUOmkiWU6lKqR8DeE0DqwiGjN/Q6Vr1z+oQ5EeSxAT4+J2xpDOhDkceyx5DpZgXOMyk7y0Y+78KdQVUouVczJFqCubMzenGR7rNzn9CHMkBpg07+14MXKqz7XQIPUB74Q8qwu44JH6rl5kGnf9bgcoJi7xVI25wse0wmvBdkGw7GQQsjq9WCX6pzf6zz40g9nnNGbAGCtgNaANX3pvGp+V5cczv2hwW9fP0/RnZd+lCyA+zT+n2qY8jCEDBogMi4+cMNHVIQ35hHkdlhHdwdFjsTJoKsuXCS6YZZD1B/GQmCPBJcHH3X+HH9pdGpU=
+    - secure: TTmBqJ0U2DrADusK6GYCGQKytnCJUqeqS+2XuE+TRp3pECdpx/arSrGit5j3oBzr2yAiakk/6N6JtWEZU7ftYaOxG7ga1Chs9Bno5T+uL67uFWkqrwxvg4AYgTUXsO+Xel7V1YmiNrXctQeIILEMB7H8OaLc6/cis3laMxY/drwexJjmf249boqfHKJuv+kHwaqfjikGO3uCLZqQsDLgzRSVH/yNvPmgUu/DZXmF6uHcfGqQRra7/VfhzFvXqAhTriYhfUyzKgouWerlcmgscN8uS9SSF8j2W+iXvR1DXyQbbf9/OV7kp+NLVTPg1wvL/RyOiQC8cMTFkdcD0lQ5xl6BBKorFPltk2Zh0Qt7syd7MJ6AMRBjoGckya6EPfTjW3rh0Pr/UDuqrVVqrP0xjgp10LUB72oW5QSDNZPXNdXgj+x9ugYtLZJIes7+yYF7ALlm7R/0RlNNiIlA6HHdOrM3Fx8BY9MHWTtqYGNxZ3yZsYZVdCk2aOyGS5o9DVXLFWWuxDFwsjokA/zc5PstIsBbyJvBqCrfoZ8YQ6b31KcdXKsN+vF+3jJP4p8nDvvz9zeIVYWGYSYrNCpRfZtLWyIyXOSDpfJ+iEgTMlxgyWuZd6dIQ3TdgcqwQhYnF5z/Nl7GgpraIvT4ttaaJkInSm6rgEA9PZr573nIWY10p28=
 before_deploy:
-- openssl aes-256-cbc -pass pass:$ENCRYPTION_PASSWORD -in secring.gpg.enc -out local.secring.gpg -d
-- openssl aes-256-cbc -pass pass:$ENCRYPTION_PASSWORD -in pubring.gpg.enc -out local.pubring.gpg -d
+  - openssl aes-256-cbc -pass pass:$ENCRYPTION_PASSWORD -in secring.gpg.enc -out local.secring.gpg -d
+  - openssl aes-256-cbc -pass pass:$ENCRYPTION_PASSWORD -in pubring.gpg.enc -out local.pubring.gpg -d
 deploy:
-- provider: script
-  script: "./publish.sh"
-  skip_cleanup: true
-  on:
-    tags: true
-    jdk: oraclejdk8
-    scala: 2.11.11
+  - provider: script
+    script: "./publish.sh"
+    skip_cleanup: true
+    on:
+      tags: true
+      jdk: openjdk8
+      scala: 2.12.11
 notifications:
   slack:
     rooms:

build.sbt

@@ -8,13 +8,13 @@ lazy val commonSettings = Seq(
   organization := "com.indix",
   organizationName := "Indix",
   organizationHomepage := Some(url("http://www.indix.com")),
-  scalaVersion := "2.11.11",
+  scalaVersion := "2.12.11",
   scalacOptions ++= Seq("-encoding", "UTF-8", "-deprecation", "-unchecked"),
   javacOptions in (Compile, compile) ++= Seq("-Xlint:deprecation", "-source", "1.8"),
   javacOptions in (Compile, doc) ++= Seq("-source", "1.8"),
   resolvers ++= Seq(
-    "Clojars" at "http://clojars.org/repo",
-    "Concurrent Maven Repo" at "http://conjars.org/repo",
+    "Clojars" at "https://clojars.org/repo",
+    "Concurrent Maven Repo" at "https://conjars.org/repo",
     "Twttr Maven Repo" at "https://maven.twttr.com/"
   )
 )
@@ -93,7 +93,7 @@ lazy val storeUtils = (project in file("util-store")).
     libraryDependencies ++= Seq(
       "org.scalatest" %% "scalatest" % "3.0.3" % Test,
       "commons-io" % "commons-io" % "2.5",
-      "com.twitter" %% "chill" % "0.8.1",
+      "com.twitter" %% "chill" % "0.9.5",
       "org.rocksdb" % "rocksdbjni" % "4.11.2"
     )
   )
@@ -104,19 +104,19 @@ lazy val sparkUtils = (project in file("util-spark")).
   settings(publishSettings: _*).
   settings(
     name := "util-spark",
-    crossScalaVersions := Seq("2.10.6", "2.11.11"),
+    crossScalaVersions := Seq("2.11.11", "2.12.11"),
     libraryDependencies ++= Seq(
       "org.scalatest" %% "scalatest" % "3.0.3" % Test,
-      "org.apache.spark" %% "spark-core" % "2.2.0",
-      "org.apache.spark" %% "spark-sql" % "2.2.0",
-      "com.databricks" %% "spark-avro" % "4.0.0",
+      "org.apache.spark" %% "spark-core" % "2.4.4",
+      "org.apache.spark" %% "spark-sql" % "2.4.4",
+      "org.apache.spark" %% "spark-avro" % "2.4.4",
       "org.apache.hadoop" % "hadoop-aws" % "2.6.0" % Test,
       "com.indix" % "dfs-datastores" % "2.0.21" excludeAll(
         ExclusionRule(organization = "org.apache.hadoop"),
         ExclusionRule(organization = "org.eclipse.jetty")
       ),
-      "org.apache.parquet" % "parquet-avro" % "1.8.1",
-      "org.bdgenomics.utils" %% "utils-misc" % "0.2.13"
+      "org.apache.parquet" % "parquet-avro" % "1.10.1",
+      "org.bdgenomics.utils" %% "utils-misc-spark2" % "0.2.16"
     )
   )
 

project/build.properties

@@ -1 +1 @@
-sbt.version = 0.13.17
+sbt.version = 0.13.18

util-spark/src/main/scala/com/indix/utils/spark/parquet/avro/ParquetAvroDataSource.scala

@@ -4,7 +4,7 @@ import java.nio.ByteBuffer
 import java.sql.Timestamp
 import java.util
 
-import com.databricks.spark.avro.SchemaConverters
+import org.apache.spark.sql.avro.SchemaConverters
 import org.apache.avro.generic.GenericData.Record
 import org.apache.avro.generic.{GenericRecord, IndexedRecord}
 import org.apache.avro.{Schema, SchemaBuilder}
@@ -23,18 +23,19 @@ import scala.reflect.ClassTag
 
 class AvroFormatter extends Serializable {
   /**
-    * This function constructs converter function for a given sparkSQL datatype. This is used in
-    * writing Avro records out to disk
-    */
+   * This function constructs converter function for a given sparkSQL datatype. This is used in
+   * writing Avro records out to disk
+   */
   def createConverterToAvro(
-                                     dataType: DataType,
-                                     structName: String,
-                                     recordNamespace: String): (Any) => Any = {
+                             dataType: DataType,
+                             structName: String,
+                             recordNamespace: String): (Any) => Any = {
     dataType match {
-      case BinaryType => (item: Any) => item match {
-        case null => null
-        case bytes: Array[Byte] => ByteBuffer.wrap(bytes)
-      }
+      case BinaryType => (item: Any) =>
+        item match {
+          case null => null
+          case bytes: Array[Byte] => ByteBuffer.wrap(bytes)
+        }
       case ByteType | ShortType | IntegerType | LongType |
            FloatType | DoubleType | StringType | BooleanType => identity
       case _: DecimalType => (item: Any) => if (item == null) null else item.toString
@@ -71,9 +72,8 @@ class AvroFormatter extends Serializable {
           }
         }
       case structType: StructType =>
-        val builder = SchemaBuilder.record(structName).namespace(recordNamespace)
-        val schema: Schema = SchemaConverters.convertStructToAvro(
-          structType, builder, recordNamespace)
+        val schema: Schema = SchemaConverters.toAvroType(
+          structType,nullable = false, structName, recordNamespace)
         val fieldConverters = structType.fields.map(field =>
           createConverterToAvro(field.dataType, field.name, recordNamespace))
         (item: Any) => {
@@ -115,7 +115,7 @@ trait ParquetAvroDataSource {
       AvroParquetOutputFormat.setSchema(job, schema)
       ParquetOutputFormat.setCompression(job, compression)
       ParquetOutputFormat.setPageSize(job, 32 * 1024 * 1024) // 128 MB seems way too large
-      val outputFormatter = if(useDFOC) classOf[ParquetAvroOutputFormatWithDFOC] else classOf[AvroParquetOutputFormat[GenericRecord]]
+      val outputFormatter = if (useDFOC) classOf[ParquetAvroOutputFormatWithDFOC] else classOf[AvroParquetOutputFormat[GenericRecord]]
       rdd.map(r => null.asInstanceOf[Void] -> r)
         .saveAsNewAPIHadoopFile(outputLocation, classOf[Void], classOf[IndexedRecord], outputFormatter, ContextUtil.getConfiguration(job))
     }
@@ -137,8 +137,7 @@ trait ParquetAvroDataSource {
 
   def toAvroSchema(schema: StructType): Schema = {
     val recordNamespace = ""
-    val build = SchemaBuilder.record("topLevelRecord").namespace(recordNamespace)
-    SchemaConverters.convertStructToAvro(schema, build, recordNamespace)
+    SchemaConverters.toAvroType(schema,nullable = false, "topLevelRecord", recordNamespace)
   }
 
 }

util-spark/src/test/scala/com/indix/utils/spark/parquet/ParquetAvroDataSourceSpec.scala

@@ -5,15 +5,30 @@ import java.io.File
 import com.google.common.io.Files
 import com.indix.utils.spark.parquet.avro.ParquetAvroDataSource
 import org.apache.commons.io.FileUtils
+import org.apache.parquet.hadoop.metadata.CompressionCodecName
 import org.apache.spark.sql.SparkSession
+import org.scalactic.Equality
+import org.scalatest.Matchers.{be, convertToAnyShouldWrapper, equal}
 import org.scalatest.{BeforeAndAfterAll, FlatSpec}
-import org.scalatest.Matchers.{be, convertToAnyShouldWrapper}
-import org.apache.parquet.hadoop.metadata.CompressionCodecName
+import java.util.{Arrays => JArrays}
 
-case class SampleAvroRecord(a: Int, b: String, c: Seq[String], d: Boolean, e: Double, f: collection.Map[String,String], g: Seq[Byte])
+case class SampleAvroRecord(a: Int, b: String, c: Seq[String], d: Boolean, e: Double, f: collection.Map[String, String], g: Array[Byte])
 
 class ParquetAvroDataSourceSpec extends FlatSpec with BeforeAndAfterAll with ParquetAvroDataSource {
   private var spark: SparkSession = _
+  implicit val sampleAvroRecordEq = new Equality[SampleAvroRecord] {
+    override def areEqual(left: SampleAvroRecord, b: Any): Boolean = b match {
+      case right: SampleAvroRecord =>
+        left.a == right.a &&
+          left.b == right.b &&
+          Equality.default[Seq[String]].areEqual(left.c, right.c) &&
+          left.d == right.d &&
+          left.e == right.e &&
+          Equality.default[collection.Map[String, String]].areEqual(left.f, right.f) &&
+          JArrays.equals(left.g, right.g)
+      case _ => false
+    }
+  }
 
   override protected def beforeAll(): Unit = {
     super.beforeAll()
@@ -33,11 +48,11 @@ class ParquetAvroDataSourceSpec extends FlatSpec with BeforeAndAfterAll with Par
     val outputLocation = Files.createTempDir().getAbsolutePath + "/output"
 
     val sampleRecords: Seq[SampleAvroRecord] = Seq(
-      SampleAvroRecord(1, "1", List("a1"), true, 1.0d, Map("a1" -> "b1"), Seq("1".toByte)),
-      SampleAvroRecord(2, "2", List("a2"), false, 2.0d, Map("a2" -> "b2"), Seq("2".toByte)),
-      SampleAvroRecord(3, "3", List("a3"), true, 3.0d, Map("a3" -> "b3"), Seq("3".toByte)),
-      SampleAvroRecord(4, "4", List("a4"), true, 4.0d, Map("a4" -> "b4"), Seq("4".toByte)),
-      SampleAvroRecord(5, "5", List("a5"), false, 5.0d, Map("a5" -> "b5"), Seq("5".toByte))
+      SampleAvroRecord(1, "1", List("a1"), true, 1.0d, Map("a1" -> "b1"), "1".getBytes),
+      SampleAvroRecord(2, "2", List("a2"), false, 2.0d, Map("a2" -> "b2"), "2".getBytes),
+      SampleAvroRecord(3, "3", List("a3"), true, 3.0d, Map("a3" -> "b3"), "3".getBytes),
+      SampleAvroRecord(4, "4", List("a4"), true, 4.0d, Map("a4" -> "b4"), "4".getBytes),
+      SampleAvroRecord(5, "5", List("a5"), false, 5.0d, Map("a5" -> "b5"), "5".getBytes)
     )
 
     val sampleDf = spark.createDataFrame(sampleRecords)
@@ -51,7 +66,9 @@ class ParquetAvroDataSourceSpec extends FlatSpec with BeforeAndAfterAll with Par
     val records: Array[SampleAvroRecord] = spark.read.parquet(outputLocation).as[SampleAvroRecord].collect()
 
     records.length should be(5)
-    records.sortBy(_.a) should be (sampleRecords.sortBy(_.a))
+    // We use === to use the custom Equality defined above for comparing Array[Byte]
+    // Ref - https://github.com/scalatest/scalatest/issues/491
+    records.sortBy(_.a) === sampleRecords.sortBy(_.a)
 
     FileUtils.deleteDirectory(new File(outputLocation))
   }