Merge master into rc

.gitignore

@@ -6,3 +6,9 @@
 /v15
 npm-debug.log
 /DataProcessing/DataX.Utilities/DataX.Utility.CodeSign/obj
+/Tests/ScenarioTester/ScenarioTester/bin
+/Tests/ScenarioTester/.vs
+*.suo
+/Tests/ScenarioTester/ScenarioTester/obj
+/Tests/ScenarioTester/ScenarioTesterTests/bin
+/Tests/ScenarioTester/ScenarioTesterTests/obj

DataProcessing/DataX.Utilities/.gitignore

@@ -0,0 +1 @@
+*/obj

DataProcessing/DataX.Utilities/DataX.Utility.CodeSign/DataX.Utility.CodeSign.csproj

@@ -2,7 +2,7 @@
 
   <PropertyGroup>
     <OutputType>Library</OutputType>
-    <TargetFramework>netcoreapp2.1</TargetFramework>
+    <TargetFramework>netcoreapp2.2</TargetFramework>
     <ApplicationIcon />
     <StartupObject />
     <SignAssembly>true</SignAssembly>

DataProcessing/Spark.nuspec

@@ -20,16 +20,21 @@
 <file src="**\azure-eventhubs-1.2.1.jar" target="lib" />
 <file src="**\azure-eventhubs-spark_2.11-2.3.6.jar" target="lib" />
 <file src="**\azure-keyvault-webkey-1.1.jar" target="lib" />
-<file src="**\datax-core_2.3_2.11-1.1.0.jar" target="lib" />
-<file src="**\datax-host_2.3_2.11-1.1.0.jar" target="lib" />
-<file src="**\datax-utility_2.3_2.11-1.1.0.jar" target="lib" />
-<file src="**\datax-keyvault_2.3_2.11-1.1.0-with-dependencies.jar" target="lib" />
-<file src="**\datax-udf-samples_2.3_2.11-1.1.0.jar" target="lib" />
+<file src="**\datax-core*.jar" target="lib" />
+<file src="**\datax-host*.jar" target="lib" />
+<file src="**\datax-utility*.jar" target="lib" />
+<file src="**\datax-keyvault*.jar" target="lib" />
+<file src="**\datax-udf-samples*.jar" target="lib" />
 <file src="**\java-uuid-generator-3.1.5.jar" target="lib" />
 <file src="**\proton-j-0.31.0.jar" target="lib" />
 <file src="**\scala-java8-compat_2.11-0.9.0.jar" target="lib" />
 <file src="**\kafka-clients-2.0.0.jar" target="lib" />
 <file src="**\spark-streaming-kafka-0-10_2.11-2.4.0.jar" target="lib" />
+<file src="**\azure-sqldb-spark-1.0.2.jar" target="lib" />
+<file src="**\hadoop-azure-2.7.3.jar" target="lib" />
+<file src="**\jetty-util-6.1.25.jar" target="lib" />
+<file src="**\json-20180813.jar" target="lib" />
+<file src="**\azure-storage-3.1.0.jar" target="lib" />
 <file src="NOTICE.txt" target="" />
   </files>
-</package>
+</package>

DataProcessing/datax-core/pom.xml

@@ -50,15 +50,15 @@ SOFTWARE
   </scm>
 
   <groupId>com.microsoft.datax</groupId>
-  <artifactId>datax-core_2.3_2.11</artifactId>
-  <version>1.1.0</version>
+  <artifactId>datax-core_2.4_2.11</artifactId>
+  <version>1.2.0</version>
   <name>Data Accelerator Core</name>
   <description>This package contains the core module of Data Accelerator functionality.</description>
   <url>https://github.com/Microsoft/data-accelerator</url>
   <packaging>jar</packaging>
 
   <properties>
-    <spark.version>2.3.0</spark.version>
+    <spark.version>2.4.0</spark.version>
     <scala.version.major>2.11</scala.version.major>
     <scala.version.minor>8</scala.version.minor>
     <scala.version>${scala.version.major}.${scala.version.minor}</scala.version>

DataProcessing/datax-core/src/main/scala/datax/constants/BlobProperties.scala

@@ -0,0 +1,10 @@
+// *********************************************************************
+// Copyright (c) Microsoft Corporation.  All rights reserved.
+// Licensed under the MIT License
+// *********************************************************************
+package datax.constants
+
+object BlobProperties {
+  // Define constants for blobs
+  val BlobHostPath = ".blob.core.windows.net"
+}

DataProcessing/datax-core/src/main/scala/datax/constants/JobArgument.scala

@@ -15,4 +15,7 @@ object JobArgument {
   def ConfName_AppInsightKeyRef = s"${ConfNamePrefix}APPINSIGHTKEYREF"
   def ConfName_BlobWriterTimeout: String = s"${ConfNamePrefix}BlobWriterTimeout"
   def ConfName_DefaultVaultName: String = s"${ConfNamePrefix}DEFAULTVAULTNAME"
+  def ConfName_DefaultStorageAccount: String = s"${ConfNamePrefix}DEFAULTSTORAGEACCOUNT"
+  def ConfName_DefaultContainer: String = s"${ConfNamePrefix}DEFAULTCONTAINER"
+  def ConfName_AzureStorageJarPath: String = s"${ConfNamePrefix}AZURESTORAGEJARPATH"
 }

DataProcessing/datax-core/src/main/scala/datax/sink/package.scala

@@ -10,7 +10,8 @@ import datax.config.SettingDictionary
 import org.apache.spark.sql.{DataFrame, Row, SparkSession}
 
 package object sink {
-  type SinkDelegate = (Row, Seq[Row], Timestamp, Int, String)=>Map[String, Int]
+  type JsonSinkDelegate = (Row, Seq[Row], Timestamp, Int, String)=>Map[String, Int]
+  type SinkDelegate = (DataFrame, Timestamp, String)=>Map[String, Int]
   type Metrics = Map[String, Double]
 
   trait SinkOperatorFactory{
@@ -21,6 +22,7 @@ package object sink {
 
   case class SinkOperator(name: String,
                           isEnabled: Boolean,
+                          sinkAsJson: Boolean,
                           flagColumnExprGenerator: () => String,
                           generator: (Int)=>SinkDelegate,
                           onInitialization: (SparkSession)=>Unit = null,

DataProcessing/datax-host/pom.xml

@@ -50,11 +50,11 @@ SOFTWARE
     </scm>
 
     <groupId>com.microsoft.datax</groupId>
-    <artifactId>datax-host_2.3_2.11</artifactId>
-    <version>1.1.0</version>
+    <artifactId>datax-host_2.4_2.11</artifactId>
+    <version>1.2.0</version>
 
     <properties>
-        <spark.version>2.3.0</spark.version>
+        <spark.version>2.4.0</spark.version>
         <scala.version.major>2.11</scala.version.major>
         <scala.version.minor>8</scala.version.minor>
         <scala.version>${scala.version.major}.${scala.version.minor}</scala.version>
@@ -117,13 +117,13 @@ SOFTWARE
         </dependency>
         <dependency>
             <groupId>com.microsoft.datax</groupId>
-            <artifactId>datax-core_2.3_2.11</artifactId>
-            <version>1.1.0</version>
+            <artifactId>datax-core_2.4_2.11</artifactId>
+            <version>1.2.0</version>
         </dependency>
         <dependency>
             <groupId>com.microsoft.datax</groupId>
-            <artifactId>datax-utility_2.3_2.11</artifactId>
-            <version>1.1.0</version>
+            <artifactId>datax-utility_2.4_2.11</artifactId>
+            <version>1.2.0</version>
         </dependency>
         <dependency>
             <groupId>com.microsoft.azure</groupId>
@@ -138,21 +138,44 @@ SOFTWARE
         <dependency>
             <groupId>com.microsoft.azure</groupId>
             <artifactId>azure-storage</artifactId>
-            <version>5.3.0</version>
+            <version>3.1.0</version>
         </dependency>
         <dependency>
             <groupId>com.microsoft.azure</groupId>
             <artifactId>azure-documentdb</artifactId>
             <version>1.16.1</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-azure</artifactId>
+            <version>2.7.3</version>
+        </dependency>
+        <dependency>
+            <groupId>org.mortbay.jetty</groupId>
+            <artifactId>jetty-util</artifactId>
+            <version>6.1.25</version>
+        </dependency>
+        <dependency>
+            <groupId>org.json</groupId>
+            <artifactId>json</artifactId>
+            <version>20180813</version>
+        </dependency>
         <dependency>
             <groupId>org.apache.kafka</groupId>
             <artifactId>kafka-clients</artifactId>
             <version>2.0.0</version>
         </dependency>
+        <dependency>
+            <groupId>com.microsoft.azure</groupId>
+            <artifactId>azure-sqldb-spark</artifactId>
+            <version>1.0.2</version>
+		    </dependency>
+		    <dependency>
+            <groupId>com.databricks</groupId>
+            <artifactId>dbutils-api_2.11</artifactId>
+            <version>0.0.3</version>
+        </dependency>
     </dependencies>
-
-
   <profiles>
     <profile>
       <id>build</id>
@@ -228,24 +251,6 @@ SOFTWARE
               </execution>
             </executions>
           </plugin>
-          <plugin>
-            <artifactId>maven-assembly-plugin</artifactId>
-            <version>2.4.1</version>
-            <configuration>
-              <descriptors>
-                <descriptor>with-dependencies.xml</descriptor>
-              </descriptors>
-            </configuration>
-            <executions>
-              <execution>
-                <id>make-assembly</id>
-                <phase>package</phase>
-                <goals>
-                  <goal>single</goal>
-                </goals>
-              </execution>
-            </executions>
-          </plugin>
           <plugin>
             <groupId>com.github.github</groupId>
             <artifactId>site-maven-plugin</artifactId>

DataProcessing/datax-host/src/main/scala/datax/app/BatchApp.scala

@@ -11,6 +11,6 @@ object BatchApp {
   def main(inputArguments: Array[String]): Unit = {
     BlobBatchingHost.runBatchApp(
       inputArguments,
-      config => CommonProcessorFactory.createProcessor(config).asBlobPointerProcessor())
+      config => CommonProcessorFactory.createProcessor(config).asBatchBlobProcessor())
   }
 }

DataProcessing/datax-host/src/main/scala/datax/client/sql/SqlConf.scala

@@ -0,0 +1,26 @@
+// *********************************************************************
+// Copyright (c) Microsoft Corporation.  All rights reserved.
+// Licensed under the MIT License
+// *********************************************************************
+package datax.client.sql
+
+case class SqlConf(name: String,
+                   connectionString: String,
+                   url: String,
+                   encrypt: String,
+                   trustServerCertificate: String,
+                   hostNameInCertificate: String,
+                   databaseName:String,
+                   table: String,
+                   writeMode: String,
+                   userName:String,
+                   password:String,
+                   filter: String,
+                   connectionTimeout: String,
+                   queryTimeout: String,
+                   useBulkCopy : Boolean,
+                   useBulkCopyTableLock: String,
+                   useBulkCopyInternalTransaction: String,
+                   bulkCopyTimeout:String,
+                   bulkCopyBatchSize:String
+                   )

DataProcessing/datax-host/src/main/scala/datax/executor/ExecutorHelper.scala

@@ -0,0 +1,85 @@
+// *********************************************************************
+// Copyright (c) Microsoft Corporation.  All rights reserved.
+// Licensed under the MIT License
+// *********************************************************************
+package datax.executor
+
+import datax.config.ConfigManager
+import datax.constants.{BlobProperties, JobArgument}
+import datax.fs.HadoopClient
+import datax.host.SparkJarLoader
+import datax.securedsetting.KeyVaultClient
+import org.apache.log4j.LogManager
+import org.apache.spark.broadcast
+import org.apache.spark.sql.SparkSession
+
+object ExecutorHelper {
+  private val logger = LogManager.getLogger(this.getClass)
+
+  /***
+    * Create broadcast variable for blob storage account key
+    * @param path blob storage path
+    * @param spark SparkSession
+    */
+  def createBlobStorageKeyBroadcastVariable(path: String, spark : SparkSession): broadcast.Broadcast[String] ={
+    addJarToExecutor(spark)
+    val sc = spark.sparkContext
+    val sa = getStorageAccountName(path)
+    var key = ""
+    KeyVaultClient.withKeyVault {vaultName => key = HadoopClient.resolveStorageAccount(vaultName, sa).get}
+    val blobStorageKey = sc.broadcast(key)
+    blobStorageKey
+  }
+
+  /***
+    * Get the storage account name from blob path
+    * @param path blob storage path
+    */
+  private def getStorageAccountName(path:String):String ={
+    val regex = s"@([a-zA-Z0-9-_]+)${BlobProperties.BlobHostPath}".r
+    regex.findFirstMatchIn(path) match {
+      case Some(partition) => partition.group(1)
+      case None =>  null
+    }
+  }
+
+  /***
+    * Add azure-storage jar to executor nodes
+    * @param spark Spark Session
+    */
+  private def addJarToExecutor(spark : SparkSession){
+    try{
+      logger.warn("Adding azure-storage jar to executor nodes")
+      withStorageAccount {(storageAccount,containerName,azureStorageJarPath) => SparkJarLoader.addJar(spark, s"wasbs://$containerName@$storageAccount${BlobProperties.BlobHostPath}$azureStorageJarPath")}
+    }
+    catch {
+      case e: Exception => {
+        logger.error(s"azure-storage jar could not be added to executer nodes", e)
+        throw e
+      }
+    }
+  }
+
+  /***
+    * a scope to execute operation with the default storageAccount/container/azureStorageJarPath, skip the operation if that doesn't exist.
+    * @param callback execution within the scope
+    */
+  private def withStorageAccount(callback: (String, String, String)=> Unit) = {
+    ConfigManager.getActiveDictionary().get(JobArgument.ConfName_DefaultStorageAccount) match {
+      case Some(storageAccount) =>
+        logger.warn(s"Default Storage Account is $storageAccount")
+        ConfigManager.getActiveDictionary().get(JobArgument.ConfName_DefaultContainer) match {
+          case Some(containerName) =>
+            logger.warn(s"Default container is $containerName")
+            ConfigManager.getActiveDictionary().get(JobArgument.ConfName_AzureStorageJarPath) match {
+              case Some(azureStorageJarPath) =>
+                logger.warn(s"Azure storage jar path is $azureStorageJarPath")
+                callback(storageAccount, containerName, azureStorageJarPath)
+              case None => logger.warn(s"No azure storage jar path is defined")
+            }
+          case None => logger.warn(s"No default container is defined")
+        }
+      case None => logger.warn(s"No default storage account is defined")
+    }
+  }
+}