/* * Copyright 2016-2020 Amazon.com, Inc. or its affiliates. All Rights Reserved. * SPDX-License-Identifier: MIT-0 */ package test.scala import com.amazonaws.services.glue.GlueContext import com.amazonaws.services.glue.schema.{Schema, TypeCode} import com.amazonaws.services.glue.schema.builders.SchemaBuilder import com.amazonaws.services.glue.util.JsonOptions import org.apache.log4j.{Level, Logger} import org.apache.spark.{SparkConf, SparkContext} import org.scalatest.{BeforeAndAfterEach, FunSuite} class DataTypeMappingTest extends FunSuite with BeforeAndAfterEach { val conf = new SparkConf().setAppName("LocalTest").setMaster("local") var spark: SparkContext = _ var glueContext: GlueContext = _ val rootLogger = Logger.getRootLogger() // Remove verbose log rootLogger.setLevel(Level.ERROR) override def beforeEach(): Unit = { spark = new SparkContext(conf) glueContext = new GlueContext(spark) } override def afterEach(): Unit = { super.afterEach() spark.stop() } test("test partner jdbc custom data type mapping") { // setup connection options var optionsMap: Map[String, Any] = Map( "query" -> "SELECT NumberOfEmployees, CreatedDate FROM Account", // use query to specify column to check "url" -> "jdbc:salesforce:user=${user};Password=${Password};SecurityToken=${SecurityToken};", "secretId"-> "secret", "className" -> "partner.jdbc.salesforce.SalesforceDriver" ) // create datasource and read data var customSource = glueContext.getSource( connectionType = "custom.jdbc", connectionOptions = JsonOptions(optionsMap), transformationContext = "") var dyf = customSource.getDynamicFrame() // check default conversion var expectedSchema = new Schema(new SchemaBuilder() .beginStruct() .atomicField("NumberOfEmployees", TypeCode.INT) .atomicField("CreatedDate", TypeCode.TIMESTAMP) .endStruct().build()) assert(dyf.schema === expectedSchema) // set up custom data type mapping from JDBC INTEGER type to Glue STRING type optionsMap = optionsMap + ("dataTypeMapping" -> Map("INTEGER" -> "STRING")) // read data again customSource = glueContext.getSource( connectionType = "custom.jdbc", connectionOptions = JsonOptions(optionsMap), transformationContext = "") dyf = customSource.getDynamicFrame() // verify JDBC INTEGER type is mapped to Glue STRING Type expectedSchema = new Schema(new SchemaBuilder() .beginStruct() .atomicField("NumberOfEmployees", TypeCode.STRING) .atomicField("CreatedDate", TypeCode.TIMESTAMP) .endStruct().build()) assert(dyf.schema === expectedSchema) } }