diff --git a/eng/versioning/external_dependencies.txt b/eng/versioning/external_dependencies.txt
index 727b19c10046..f3eb3b7ef2c6 100644
--- a/eng/versioning/external_dependencies.txt
+++ b/eng/versioning/external_dependencies.txt
@@ -286,6 +286,9 @@ cosmos_org.scala-lang:scala-library;2.12.10
 
 # Cosmos Spark connector tests only
 cosmos_org.scalatest:scalatest_2.12;3.2.2
+cosmos_org.scalatest:scalatest-flatspec_2.12;3.2.3
+cosmos_org.scalactic:scalactic_2.12;3.2.3
+cosmos_org.scalamock:scalamock_2.12;5.0.0
 
 # Maven Tools for Cosmos Spark connector only
 cosmos_net.alchim31.maven:scala-maven-plugin;4.4.0
diff --git a/sdk/cosmos/azure-cosmos-spark_3-0_2-12/pom.xml b/sdk/cosmos/azure-cosmos-spark_3-0_2-12/pom.xml
index 79378f5cc27e..84309b1034e6 100644
--- a/sdk/cosmos/azure-cosmos-spark_3-0_2-12/pom.xml
+++ b/sdk/cosmos/azure-cosmos-spark_3-0_2-12/pom.xml
@@ -63,6 +63,12 @@
       <version>1.7.30</version> <!-- {x-version-update;org.slf4j:slf4j-simple;external_dependency} -->
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.assertj</groupId>
+      <artifactId>assertj-core</artifactId>
+      <version>3.16.1</version> <!-- {x-version-update;org.assertj:assertj-core;external_dependency} -->
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>org.scala-lang</groupId>
       <artifactId>scala-library</artifactId>
@@ -81,6 +87,25 @@
       <version>3.2.2</version> <!-- {x-version-update;cosmos_org.scalatest:scalatest_2.12;external_dependency} -->
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.scalatest</groupId>
+      <artifactId>scalatest-flatspec_2.12</artifactId>
+      <version>3.2.3</version> <!-- {x-version-update;cosmos_org.scalatest:scalatest-flatspec_2.12;external_dependency} -->
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.scalactic</groupId>
+      <artifactId>scalactic_2.12</artifactId>
+      <version>3.2.3</version> <!-- {x-version-update;cosmos_org.scalactic:scalactic_2.12;external_dependency} -->
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.scalamock</groupId>
+      <artifactId>scalamock_2.12</artifactId>
+      <version>5.0.0</version> <!-- {x-version-update;cosmos_org.scalamock:scalamock_2.12;external_dependency} -->
+      <scope>test</scope>
+    </dependency>
+
     <!-- Added this provided dependency to include necessary annotations used by "reactor-core".
          Without this dependency, javadoc throws a warning as it cannot find enum When.MAYBE
          which is used in @Nullable annotation in reactor core classes.
@@ -108,6 +133,7 @@
         <filtering>true</filtering>
         <includes>
           <include>META-INF/project.properties</include>
+          <include>META-INF/services/org.apache.spark.sql.sources.DataSourceRegister</include>
         </includes>
       </resource>
     </resources>
diff --git a/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/main/java/module-info.java b/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/main/java/module-info.java
deleted file mode 100644
index e47c0ae96cff..000000000000
--- a/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/main/java/module-info.java
+++ /dev/null
@@ -1,6 +0,0 @@
-// Copyright (c) Microsoft Corporation. All rights reserved.
-// Licensed under the MIT License.
-
-module com.azure.cosmos.spark {
-    requires transitive com.azure.cosmos;
-}
\ No newline at end of file
diff --git a/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister b/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister
new file mode 100644
index 000000000000..c5c028b9f3bf
--- /dev/null
+++ b/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister
@@ -0,0 +1 @@
+com.azure.cosmos.spark.CosmosItemsDataSource
diff --git a/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/main/scala/com/azure/cosmos/spark/CosmosBatchWriter.scala b/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/main/scala/com/azure/cosmos/spark/CosmosBatchWriter.scala
new file mode 100644
index 000000000000..2b0f64c246ca
--- /dev/null
+++ b/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/main/scala/com/azure/cosmos/spark/CosmosBatchWriter.scala
@@ -0,0 +1,19 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+package com.azure.cosmos.spark
+
+import org.apache.spark.sql.connector.write.{BatchWrite, DataWriterFactory, PhysicalWriteInfo, WriterCommitMessage}
+
+class CosmosBatchWriter extends BatchWrite with CosmosLoggingTrait {
+  logInfo(s"Instantiated ${this.getClass.getSimpleName}")
+
+  override def createBatchWriterFactory(physicalWriteInfo: PhysicalWriteInfo): DataWriterFactory = new CosmosDataWriteFactory()
+
+  override def commit(writerCommitMessages: Array[WriterCommitMessage]): Unit = {
+    // TODO
+  }
+
+  override def abort(writerCommitMessages: Array[WriterCommitMessage]): Unit = {
+    // TODO
+  }
+}
diff --git a/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/main/scala/com/azure/cosmos/spark/CosmosDataWriteFactory.scala b/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/main/scala/com/azure/cosmos/spark/CosmosDataWriteFactory.scala
new file mode 100644
index 000000000000..b550060567cd
--- /dev/null
+++ b/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/main/scala/com/azure/cosmos/spark/CosmosDataWriteFactory.scala
@@ -0,0 +1,57 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+package com.azure.cosmos.spark
+
+import java.util.UUID
+
+import com.azure.cosmos.implementation.TestConfigurations
+import com.azure.cosmos.{ConsistencyLevel, CosmosClientBuilder}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.connector.write.{DataWriter, DataWriterFactory, WriterCommitMessage}
+import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
+
+class CosmosDataWriteFactory extends DataWriterFactory with CosmosLoggingTrait {
+  logInfo(s"Instantiated ${this.getClass.getSimpleName}")
+
+  override def createWriter(i: Int, l: Long): DataWriter[InternalRow] = new CosmosWriter()
+
+  class CosmosWriter() extends DataWriter[InternalRow] {
+    logInfo(s"Instantiated ${this.getClass.getSimpleName}")
+
+    // TODO moderakh account config and databaseName, containerName need to passed down from the user
+    val client = new CosmosClientBuilder()
+      .key(TestConfigurations.MASTER_KEY)
+      .endpoint(TestConfigurations.HOST)
+      .consistencyLevel(ConsistencyLevel.EVENTUAL)
+      .buildAsyncClient();
+    val databaseName = "testDB"
+    val containerName = "testContainer"
+
+    override def write(internalRow: InternalRow): Unit = {
+      // TODO moderakh: schema is hard coded for now to make end to end TestE2EMain work implement schema inference code
+      val userProvidedSchema = StructType(Seq(StructField("number", IntegerType), StructField("word", StringType)))
+
+      val objectNode = CosmosRowConverter.internalRowToObjectNode(internalRow, userProvidedSchema)
+      // TODO: moderakh how should we handle absence of id?
+      if (!objectNode.has("id")) {
+        objectNode.put("id", UUID.randomUUID().toString)
+      }
+      client.getDatabase(databaseName)
+        .getContainer(containerName)
+        .createItem(objectNode)
+        .block()
+    }
+
+    override def commit(): WriterCommitMessage = {
+      new WriterCommitMessage {}
+    }
+
+    override def abort(): Unit = {
+      // TODO
+    }
+
+    override def close(): Unit = {
+      // TODO
+    }
+  }
+}
diff --git a/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/main/scala/com/azure/cosmos/spark/CosmosItemsDataSource.scala b/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/main/scala/com/azure/cosmos/spark/CosmosItemsDataSource.scala
new file mode 100644
index 000000000000..dbd3c22bbaf0
--- /dev/null
+++ b/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/main/scala/com/azure/cosmos/spark/CosmosItemsDataSource.scala
@@ -0,0 +1,30 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+package com.azure.cosmos.spark
+
+import java.util
+
+import org.apache.spark.sql.connector.catalog.{Table, TableProvider}
+import org.apache.spark.sql.connector.expressions.Transform
+import org.apache.spark.sql.sources.DataSourceRegister
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+class CosmosItemsDataSource extends DataSourceRegister with TableProvider with CosmosLoggingTrait {
+  logInfo(s"Instantiated ${this.getClass.getSimpleName}")
+
+  override def inferSchema(caseInsensitiveStringMap: CaseInsensitiveStringMap): StructType = {
+    // scalastyle:off null
+    getTable(null,
+      Array.empty[Transform],
+      caseInsensitiveStringMap.asCaseSensitiveMap()).schema()
+    // scalastyle:on null
+  }
+
+  override def shortName(): String = "cosmos.items"
+
+  override def getTable(structType: StructType, transforms: Array[Transform], map: util.Map[String, String]): Table = {
+    // getTable - This is used for loading table with user specified schema and other transformations.
+    new CosmosTable(structType, transforms, map)
+  }
+}
diff --git a/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/main/scala/com/azure/cosmos/spark/CosmosLoggingTrait.scala b/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/main/scala/com/azure/cosmos/spark/CosmosLoggingTrait.scala
new file mode 100644
index 000000000000..a44ffd40fb93
--- /dev/null
+++ b/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/main/scala/com/azure/cosmos/spark/CosmosLoggingTrait.scala
@@ -0,0 +1,59 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+package com.azure.cosmos.spark
+
+import org.slf4j.{Logger, LoggerFactory}
+
+trait CosmosLoggingTrait {
+  // Make the log field transient so that objects with Logging can
+  // be serialized and used on another machine
+  @transient private lazy val log : Logger = LoggerFactory.getLogger(logName)
+
+  // Method to get the logger name for this object
+  protected def logName: String = {
+    // Ignore trailing $'s in the class names for Scala objects
+    this.getClass.getName.stripSuffix("$")
+  }
+
+  // Log methods that take only a String
+  protected def logInfo(msg: => String) {
+    if (log.isInfoEnabled) log.info(msg)
+  }
+
+  protected def logDebug(msg: => String) {
+    if (log.isDebugEnabled) log.debug(msg)
+  }
+
+  protected def logTrace(msg: => String) {
+    if (log.isTraceEnabled) log.trace(msg)
+  }
+
+  protected def logWarning(msg: => String) {
+    if (log.isWarnEnabled) log.warn(msg)
+  }
+
+  protected def logError(msg: => String) {
+    if (log.isErrorEnabled) log.error(msg)
+  }
+
+  // Log methods that take Throwables (Exceptions/Errors) too
+  protected def logInfo(msg: => String, throwable: Throwable) {
+    if (log.isInfoEnabled) log.info(msg, throwable)
+  }
+
+  protected def logDebug(msg: => String, throwable: Throwable) {
+    if (log.isDebugEnabled) log.debug(msg, throwable)
+  }
+
+  protected def logTrace(msg: => String, throwable: Throwable) {
+    if (log.isTraceEnabled) log.trace(msg, throwable)
+  }
+
+  protected def logWarning(msg: => String, throwable: Throwable) {
+    if (log.isWarnEnabled) log.warn(msg, throwable)
+  }
+
+  protected def logError(msg: => String, throwable: Throwable) {
+    if (log.isErrorEnabled) log.error(msg, throwable)
+  }
+}
diff --git a/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/main/scala/com/azure/cosmos/spark/CosmosTable.scala b/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/main/scala/com/azure/cosmos/spark/CosmosTable.scala
new file mode 100644
index 000000000000..99d60a31d20a
--- /dev/null
+++ b/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/main/scala/com/azure/cosmos/spark/CosmosTable.scala
@@ -0,0 +1,40 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+package com.azure.cosmos.spark
+
+import java.util
+
+import org.apache.spark.sql.connector.catalog.{SupportsWrite, Table, TableCapability}
+import org.apache.spark.sql.connector.expressions.Transform
+import org.apache.spark.sql.connector.write.{LogicalWriteInfo, WriteBuilder}
+import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
+
+// scalastyle:off underscore.import
+import scala.collection.JavaConverters._
+// scalastyle:on underscore.import
+
+/**
+ * CosmosTable is the entry point this is registered in the spark
+ * @param userProvidedSchema
+ * @param transforms
+ * @param map
+ */
+class CosmosTable(val userProvidedSchema: StructType,
+                  val transforms: Array[Transform],
+                  val map: util.Map[String, String])
+  extends Table with SupportsWrite with CosmosLoggingTrait {
+  logInfo(s"Instantiated ${this.getClass.getSimpleName}")
+
+  override def name(): String = "com.azure.cosmos.spark.write"
+
+  override def schema(): StructType = {
+    // TODO: moderakh add support for schema inference
+    // for now schema is hard coded to make TestE2EMain to work
+    StructType(Seq(StructField("number", IntegerType), StructField("word", StringType)))
+  }
+
+  override def capabilities(): util.Set[TableCapability] = Set(TableCapability.BATCH_WRITE).asJava
+
+  override def newWriteBuilder(logicalWriteInfo: LogicalWriteInfo): WriteBuilder = new CosmosWriterBuilder
+
+}
diff --git a/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/main/scala/com/azure/cosmos/spark/CosmosWriterBuilder.scala b/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/main/scala/com/azure/cosmos/spark/CosmosWriterBuilder.scala
new file mode 100644
index 000000000000..fd0ef32d9dda
--- /dev/null
+++ b/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/main/scala/com/azure/cosmos/spark/CosmosWriterBuilder.scala
@@ -0,0 +1,11 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+package com.azure.cosmos.spark
+
+import org.apache.spark.sql.connector.write.{BatchWrite, WriteBuilder}
+
+class CosmosWriterBuilder extends WriteBuilder with CosmosLoggingTrait {
+  logInfo(s"Instantiated ${this.getClass.getSimpleName}")
+
+  override def buildForBatch(): BatchWrite = new CosmosBatchWriter()
+}
diff --git a/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/main/scala/com/azure/cosmos/spark/Dummy.scala b/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/main/scala/com/azure/cosmos/spark/Dummy.scala
deleted file mode 100644
index 9bac3a1b309f..000000000000
--- a/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/main/scala/com/azure/cosmos/spark/Dummy.scala
+++ /dev/null
@@ -1,9 +0,0 @@
-// Copyright (c) Microsoft Corporation. All rights reserved.
-// Licensed under the MIT License.
-package com.azure.cosmos.spark
-
-class Dummy {
-  def sayHello(text: String): String = {
-    "Hello " + text
-  }
-}
diff --git a/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/main/scala/com/azure/cosmos/spark/RowConverter.scala b/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/main/scala/com/azure/cosmos/spark/RowConverter.scala
new file mode 100644
index 000000000000..d27a3290ba41
--- /dev/null
+++ b/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/main/scala/com/azure/cosmos/spark/RowConverter.scala
@@ -0,0 +1,157 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+package com.azure.cosmos.spark
+
+import java.sql.{Date, Timestamp}
+
+import com.fasterxml.jackson.databind.node.{ArrayNode, ObjectNode}
+import com.fasterxml.jackson.databind.{JsonNode, ObjectMapper}
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.UnsafeMapData
+import org.apache.spark.sql.catalyst.util.ArrayData
+// scalastyle:off underscore.import
+import org.apache.spark.sql.types._
+// scalastyle:on underscore.import
+import org.apache.spark.unsafe.types.UTF8String
+
+
+// TODO: moderakh more discussion is required to decide how to do row conversion
+// see https://github.com/Azure/azure-sdk-for-java/pull/17532#discussion_r522749612 for more info
+
+// TODO add more unit tests for this class to CosmosRowConverterSpec.
+
+// scalastyle:off multiple.string.literals
+object CosmosRowConverter
+  extends Serializable
+    with CosmosLoggingTrait {
+
+  // TODO moderakh make this configurable
+  val objectMapper = new ObjectMapper();
+
+  def rowToObjectNode(row: Row): ObjectNode = {
+    val objectNode: ObjectNode = objectMapper.createObjectNode();
+    row.schema.fields.zipWithIndex.foreach({
+      case (field, i) => {
+        objectNode.set(field.name, convertToJson(row.get(i), field.dataType, isInternalRow = false))
+        objectNode
+      }
+    })
+    objectNode
+  }
+
+  def internalRowToObjectNode(internalRow: InternalRow, schema: StructType): ObjectNode = {
+    val objectNode: ObjectNode = objectMapper.createObjectNode();
+    schema.fields.zipWithIndex.foreach({
+      case (field, i) => {
+        objectNode.set(field.name, convertToJson(internalRow.get(i, field.dataType), field.dataType, isInternalRow = true))
+        objectNode
+      }
+    })
+    objectNode
+  }
+
+  // scalastyle:off cyclomatic.complexity
+  private def convertToJson(element: Any, elementType: DataType, isInternalRow: Boolean): JsonNode = {
+    elementType match {
+      case BinaryType => objectMapper.convertValue(element.asInstanceOf[Array[Byte]], classOf[JsonNode])
+      case BooleanType => objectMapper.convertValue(element.asInstanceOf[Boolean], classOf[JsonNode])
+      case DateType => objectMapper.convertValue(element.asInstanceOf[Date].getTime, classOf[JsonNode])
+      case DoubleType => objectMapper.convertValue(element.asInstanceOf[Double], classOf[JsonNode])
+      case IntegerType => objectMapper.convertValue(element.asInstanceOf[Int], classOf[JsonNode])
+      case LongType => objectMapper.convertValue(element.asInstanceOf[Long], classOf[JsonNode])
+      case FloatType => objectMapper.convertValue(element.asInstanceOf[Float], classOf[JsonNode])
+      case NullType => objectMapper.nullNode()
+      case DecimalType() => if (element.isInstanceOf[Decimal]) {
+        objectMapper.convertValue(element.asInstanceOf[Decimal].toJavaBigDecimal, classOf[JsonNode])
+      } else if (element.isInstanceOf[java.lang.Long]) {
+        objectMapper.convertValue(new java.math.BigDecimal(element.asInstanceOf[java.lang.Long]), classOf[JsonNode])
+      } else {
+        objectMapper.convertValue(element.asInstanceOf[java.math.BigDecimal], classOf[JsonNode])
+      }
+      case StringType =>
+        if (isInternalRow) {
+          objectMapper.convertValue(element.asInstanceOf[UTF8String].toString, classOf[JsonNode])
+        } else {
+          objectMapper.convertValue(element.asInstanceOf[String], classOf[JsonNode])
+        }
+      case TimestampType => if (element.isInstanceOf[java.lang.Long]) {
+        objectMapper.convertValue(element.asInstanceOf[java.lang.Long], classOf[JsonNode])
+      } else {
+        objectMapper.convertValue(element.asInstanceOf[Timestamp].getTime, classOf[JsonNode])
+      }
+      case arrayType: ArrayType => arrayTypeRouterToJsonArray(arrayType.elementType, element, isInternalRow)
+
+      case mapType: MapType if isInternalRow =>
+        mapType.keyType match {
+          case StringType =>
+            // convert from UnsafeMapData to scala Map
+            val unsafeMap = element.asInstanceOf[UnsafeMapData]
+            val keys: Array[String] = unsafeMap.keyArray().toArray[UTF8String](StringType).map(_.toString)
+            val values: Array[AnyRef] = unsafeMap.valueArray().toObjectArray(mapType.valueType)
+            mapTypeToObjectNode(mapType.valueType, keys.zip(values).toMap, isInternalRow)
+          case _ => throw new Exception(
+            s"Cannot cast $element into a Json value. MapTypes must have keys of StringType for conversion into a Document"
+          )
+        }
+
+      case structType: StructType => rowTypeRouterToJsonArray(element, structType)
+      case _ =>
+        throw new Exception(s"Cannot cast $element into a Json value. $elementType has no matching Json value.")
+    }
+  }
+  // scalastyle:off cyclomatic.complexity
+
+  private def mapTypeToObjectNode(valueType: DataType, data: Map[String, Any], isInternalRow: Boolean): ObjectNode = {
+    val jsonObject: ObjectNode = objectMapper.createObjectNode();
+    valueType match {
+      case subDocuments: StructType => data.map(kv => jsonObject.put(kv._1, rowTypeRouterToJsonArray(kv._2, subDocuments)))
+      case subArray: ArrayType => data.map(kv => jsonObject.put(kv._1, arrayTypeRouterToJsonArray(subArray.elementType, kv._2, isInternalRow)))
+      case _ => data.map(kv => jsonObject.put(kv._1, convertToJson(kv._2, valueType, isInternalRow)))
+    }
+    jsonObject
+  }
+
+  private def arrayTypeRouterToJsonArray(elementType: DataType, data: Any, isInternalRow: Boolean): ArrayNode = {
+    data match {
+      case d: Seq[_] => arrayTypeToJsonArray(elementType, d, isInternalRow)
+      case d: ArrayData => arrayDataTypeToJsonArray(elementType, d, isInternalRow)
+      case _ => throw new Exception(s"Cannot cast $data into a Json value. ArrayType $elementType has no matching Json value.")
+    }
+  }
+
+  private def arrayTypeToJsonArray(elementType: DataType, data: Seq[Any], isInternalRow: Boolean): ArrayNode = {
+    val arrayNode = objectMapper.createArrayNode()
+
+    elementType match {
+      case subDocuments: StructType => data.foreach(x => arrayNode.add(rowTypeRouterToJsonArray(x, subDocuments)))
+      case subArray: ArrayType => data.foreach(x => arrayNode.add(arrayTypeRouterToJsonArray(subArray.elementType, x, isInternalRow)))
+      case _ => data.foreach(x => arrayNode.add(convertToJson(x, elementType, isInternalRow)))
+    }
+    // When constructing the JSONArray, the internalData should contain JSON-compatible objects in order for the schema to be mantained.
+    // Otherwise, the data will be converted into String.
+    arrayNode
+  }
+
+  private def arrayDataTypeToJsonArray(elementType: DataType, data: ArrayData, isInternalRow: Boolean): ArrayNode = {
+    val arrayNode = objectMapper.createArrayNode()
+
+    elementType match {
+      case subDocuments: StructType => data.foreach(elementType, (_, x) => arrayNode.add(rowTypeRouterToJsonArray(x, subDocuments)))
+      case subArray: ArrayType => data.foreach(elementType, (_, x) => arrayNode.add(arrayTypeRouterToJsonArray(subArray.elementType, x, isInternalRow)))
+      case _ => data.foreach(elementType, (_, x) => arrayNode.add(convertToJson(x, elementType, isInternalRow)))
+    }
+    // When constructing the JSONArray, the internalData should contain JSON-compatible objects in order for the schema to be mantained.
+    // Otherwise, the data will be converted into String.
+    arrayNode
+  }
+
+  private def rowTypeRouterToJsonArray(element: Any, schema: StructType) : ObjectNode = {
+    element match {
+      case e: Row => rowToObjectNode(e)
+      case e: InternalRow => internalRowToObjectNode(e, schema)
+      case _ => throw new Exception(s"Cannot cast $element into a Json value. Struct $element has no matching Json value.")
+    }
+  }
+}
+// scalastyle:on multiple.string.literals
diff --git a/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/test/scala/com/azure/cosmos/spark/CosmosRowConverterSpec.scala b/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/test/scala/com/azure/cosmos/spark/CosmosRowConverterSpec.scala
new file mode 100644
index 000000000000..55d100c7841c
--- /dev/null
+++ b/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/test/scala/com/azure/cosmos/spark/CosmosRowConverterSpec.scala
@@ -0,0 +1,66 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+package com.azure.cosmos.spark
+
+import com.fasterxml.jackson.databind.node.ArrayNode
+import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema
+import org.apache.spark.sql.types.{ArrayType, IntegerType, NullType, StringType, StructField, StructType}
+import org.assertj.core.api.Assertions.assertThat
+
+class CosmosRowConverterSpec extends UnitSpec {
+  //scalastyle:off null
+  //scalastyle:off multiple.string.literals
+
+  "basic spark row" should "translate to ObjectNode" in {
+
+    val colName1 = "testCol1"
+    val colName2 = "testCol2"
+    val colVal1 = 8
+    val colVal2 = "strVal"
+
+    val row = new GenericRowWithSchema(
+      Array(colVal1, colVal2),
+      StructType(Seq(StructField(colName1, IntegerType), StructField(colName2, StringType))))
+
+    val objectNode = CosmosRowConverter.rowToObjectNode(row)
+    assertThat(objectNode.get(colName1).asInt()).isEqualTo(colVal1)
+    assertThat(objectNode.get(colName2).asText()).isEqualTo(colVal2)
+  }
+
+  "null type in spark row" should "translate to null in ObjectNode" in {
+
+    val colName1 = "testCol1"
+    val colName2 = "testCol2"
+    val colVal1 = null
+    val colVal2 = "strVal"
+
+    val row = new GenericRowWithSchema(
+      Array(colVal1, colVal2),
+      StructType(Seq(StructField(colName1, NullType), StructField(colName2, StringType))))
+
+    val objectNode = CosmosRowConverter.rowToObjectNode(row)
+    assertThat(objectNode.get(colName1).isNull).isTrue
+    assertThat(objectNode.get(colName2).asText()).isEqualTo(colVal2)
+  }
+
+  "array in spark row" should "translate to null in ArrayNode" in {
+    val colName1 = "testCol1"
+    val colName2 = "testCol2"
+    val colVal1 = "strVal"
+
+    val row = new GenericRowWithSchema(
+      Array(Seq("arrayElement1", "arrayElement2"), colVal1),
+      StructType(Seq(StructField(colName1, ArrayType(StringType, true)), StructField(colName2, StringType))))
+
+    val objectNode = CosmosRowConverter.rowToObjectNode(row)
+    assertThat(objectNode.get(colName1).isArray)
+    assertThat(objectNode.get(colName1).asInstanceOf[ArrayNode]).hasSize(2)
+    assertThat(objectNode.get(colName1).asInstanceOf[ArrayNode].get(0).asText()).isEqualTo("arrayElement1")
+    assertThat(objectNode.get(colName1).asInstanceOf[ArrayNode].get(1).asText()).isEqualTo("arrayElement2")
+
+    assertThat(objectNode.get(colName2).asText()).isEqualTo(colVal1)
+  }
+  //scalastyle:on null
+  //scalastyle:on multiple.string.literals
+  // TODO moderakh add more tests for all primitive types, Map, List, Nested Type, ...
+}
diff --git a/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/test/scala/com/azure/cosmos/spark/TestE2EMain.scala b/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/test/scala/com/azure/cosmos/spark/TestE2EMain.scala
new file mode 100644
index 000000000000..9a56eca1c446
--- /dev/null
+++ b/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/test/scala/com/azure/cosmos/spark/TestE2EMain.scala
@@ -0,0 +1,55 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+package com.azure.cosmos.spark
+
+import com.azure.cosmos.implementation.TestConfigurations
+import com.azure.cosmos.{ConsistencyLevel, CosmosClientBuilder}
+import org.apache.spark.sql.SparkSession
+
+object TestE2EMain {
+  def main(args: Array[String]) {
+    val cosmosEndpoint = TestConfigurations.HOST
+    val cosmosMasterKey = TestConfigurations.MASTER_KEY
+    val cosmosDatabase = "testDB"
+    val cosmosContainer = "testContainer"
+
+    val client = new CosmosClientBuilder()
+      .endpoint(cosmosEndpoint)
+      .key(cosmosMasterKey)
+      .consistencyLevel(ConsistencyLevel.EVENTUAL)
+      .buildAsyncClient()
+
+    client.createDatabaseIfNotExists(cosmosDatabase).block()
+    client.getDatabase(cosmosDatabase).createContainerIfNotExists(cosmosContainer, "/id").block()
+    client.close()
+
+    val cfg = Map("spark.cosmos.accountEndpoint" -> cosmosEndpoint,
+      "spark.cosmos.accountKey" -> cosmosMasterKey,
+      "spark.cosmos.database" -> cosmosDatabase,
+      "spark.cosmos.container" -> cosmosContainer
+    )
+
+    val spark = SparkSession.builder()
+      .appName("spark connector sample")
+      .master("local")
+      .getOrCreate()
+
+    // scalastyle:off underscore.import
+    // scalastyle:off import.grouping
+    import spark.implicits._
+    // scalastyle:on underscore.import
+    // scalastyle:on import.grouping
+
+    val df = Seq(
+      (8, "bat"),
+      (64, "mouse"),
+      (1, "horse")
+    ).toDF("number", "word")
+    df.printSchema()
+
+    df.write.format("cosmos.items").mode("append").options(cfg).save()
+    df.show()
+
+    spark.close()
+  }
+}
diff --git a/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/test/scala/com/azure/cosmos/spark/UnitSpec.scala b/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/test/scala/com/azure/cosmos/spark/UnitSpec.scala
new file mode 100644
index 000000000000..56a0fa8fd6be
--- /dev/null
+++ b/sdk/cosmos/azure-cosmos-spark_3-0_2-12/src/test/scala/com/azure/cosmos/spark/UnitSpec.scala
@@ -0,0 +1,16 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.cosmos.spark
+
+import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach, Inside, Inspectors, OptionValues}
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+abstract class UnitSpec extends AnyFlatSpec
+  with BeforeAndAfterAll
+  with BeforeAndAfterEach
+  with Matchers
+  with OptionValues
+  with Inside
+  with Inspectors