allow authentication via access token in addition to username/password #16
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -17,7 +17,7 @@ package com.springml.spark.salesforce | |
|
|
||
| import java.text.SimpleDateFormat | ||
|
|
||
| import com.springml.salesforce.wave.api.{APIFactory, BulkAPI, ForceAPI, WaveAPI} | ||
| import org.apache.log4j.Logger | ||
| import org.apache.spark.sql.sources.{BaseRelation, CreatableRelationProvider, RelationProvider, SchemaRelationProvider} | ||
| import org.apache.spark.sql.types.StructType | ||
|
|
@@ -50,9 +50,19 @@ class DefaultSource extends RelationProvider with SchemaRelationProvider with Cr | |
| * | ||
| */ | ||
| override def createRelation(sqlContext: SQLContext, parameters: Map[String, String], schema: StructType) = { | ||
| val username = optionalParam(parameters, "SF_USERNAME", "username") | ||
| val authToken = optionalParam(parameters, "SF_AUTHTOKEN", "authToken") | ||
| validateMutualExclusive(username, authToken, "username", "authToken") | ||
| var password = "" | ||
| var serverUrl = "" | ||
| if (username.isDefined) { | ||
| password = param(parameters, "SF_PASSWORD", "password") | ||
| serverUrl = parameters.getOrElse("login", "https://login.salesforce.com") | ||
| } else { | ||
| serverUrl = parameters.getOrElse("instanceUrl", "https://login.salesforce.com") | ||
|
There was a problem hiding this comment. Why don't we use the existing property for instanceUrl? |
||
| } | ||
|
|
||
|
|
||
| val version = parameters.getOrElse("version", "36.0") | ||
| val saql = parameters.get("saql") | ||
| val soql = parameters.get("soql") | ||
|
|
@@ -71,17 +81,30 @@ class DefaultSource extends RelationProvider with SchemaRelationProvider with Cr | |
| val inferSchemaFlag = flag(inferSchema, "inferSchema") | ||
|
|
||
| if (saql.isDefined) { | ||
|
|
||
| var waveAPI:WaveAPI = null | ||
|
|
||
| if (username.isDefined) { | ||
| waveAPI = APIFactory.getInstance.waveAPI(username.get, password, serverUrl, version) | ||
| } else { | ||
| waveAPI = APIFactory.getInstance.waveAPIwAuthToken(authToken.get, serverUrl, version) | ||
|
There was a problem hiding this comment. We need PR for salesforce-wave-api as well |
||
| } | ||
|
|
||
| DatasetRelation(waveAPI, null, saql.get, schema, sqlContext, | ||
| resultVariable, pageSize.toInt, sampleSize.toInt, | ||
| encodeFields, inferSchemaFlag, replaceDatasetNameWithId.toBoolean, sdf(dateFormat)) | ||
| } else { | ||
| if (replaceDatasetNameWithId.toBoolean) { | ||
| logger.warn("Ignoring 'replaceDatasetNameWithId' option as it is not applicable to soql") | ||
| } | ||
| var forceAPI: ForceAPI = null | ||
| if (username.isDefined) { | ||
| forceAPI = APIFactory.getInstance.forceAPI(username.get, password, serverUrl, | ||
| version, Integer.getInteger(pageSize), Integer.getInteger(maxRetry)) | ||
| } else { | ||
| forceAPI = APIFactory.getInstance.forceAPIwAuthToken(authToken.get, serverUrl, | ||
| version, Integer.getInteger(pageSize), Integer.getInteger(maxRetry)) | ||
| } | ||
| DatasetRelation(null, forceAPI, soql.get, schema, sqlContext, | ||
| null, 0, sampleSize.toInt, encodeFields, inferSchemaFlag, | ||
| replaceDatasetNameWithId.toBoolean, sdf(dateFormat)) | ||
|
|
@@ -91,12 +114,23 @@ class DefaultSource extends RelationProvider with SchemaRelationProvider with Cr | |
|
|
||
| override def createRelation(sqlContext: SQLContext, mode: SaveMode, parameters: Map[String, String], data: DataFrame): BaseRelation = { | ||
|
|
||
| val username = optionalParam(parameters, "SF_USERNAME", "username") | ||
| val authToken = optionalParam(parameters, "SF_AUTHTOKEN", "authToken") | ||
|
|
||
| validateMutualExclusive(username, authToken, "username", "authToken") | ||
|
|
||
| var password = "" | ||
| var serverUrl = "" | ||
| if (username.isDefined) { | ||
| password = param(parameters, "SF_PASSWORD", "password") | ||
| serverUrl = parameters.getOrElse("login", "https://login.salesforce.com") | ||
| } else { | ||
| serverUrl = parameters.getOrElse("instanceUrl", "https://login.salesforce.com") | ||
| } | ||
|
|
||
| val datasetName = parameters.get("datasetName") | ||
| val sfObject = parameters.get("sfObject") | ||
| val appName = parameters.getOrElse("appName", null) | ||
| val version = parameters.getOrElse("version", "36.0") | ||
| val usersMetadataConfig = parameters.get("metadataConfig") | ||
| val upsert = parameters.getOrElse("upsert", "false") | ||
|
|
@@ -115,21 +149,22 @@ class DefaultSource extends RelationProvider with SchemaRelationProvider with Cr | |
| } | ||
|
|
||
| logger.info("Writing dataframe into Salesforce Wave") | ||
| writeInSalesforceWave(username, password, authToken, serverUrl, version, | ||
| datasetName.get, appName, usersMetadataConfig, mode, | ||
| flag(upsert, "upsert"), flag(monitorJob, "monitorJob"), data, metadataFile) | ||
| } else { | ||
| logger.info("Updating Salesforce Object") | ||
| updateSalesforceObject(username, password, authToken, serverUrl, version, sfObject.get, mode, data) | ||
| } | ||
|
|
||
| return createReturnRelation(data) | ||
| } | ||
|
|
||
| private def updateSalesforceObject( | ||
| username: Option[String], | ||
| password: String, | ||
| authToken: Option[String], | ||
| serverUrl: String, | ||
| version: String, | ||
| sfObject: String, | ||
| mode: SaveMode, | ||
|
|
@@ -141,8 +176,13 @@ class DefaultSource extends RelationProvider with SchemaRelationProvider with Cr | |
| val repartitionedRDD = Utils.repartition(data.rdd) | ||
| logger.info("no of partitions after repartitioning is " + repartitionedRDD.partitions.length) | ||
|
|
||
| var bulkAPI:BulkAPI = null | ||
| if (username.isDefined) { | ||
| bulkAPI = APIFactory.getInstance.bulkAPI(username.get, password, serverUrl, version) | ||
| } else { | ||
| bulkAPI = APIFactory.getInstance.bulkAPIwAuthToken(authToken.get, serverUrl, version) | ||
| } | ||
| val writer = new SFObjectWriter(username, password, authToken, serverUrl, version, sfObject, mode, csvHeader) | ||
| logger.info("Writing data") | ||
| val successfulWrite = writer.writeData(repartitionedRDD) | ||
| logger.info(s"Writing data was successful was $successfulWrite") | ||
|
|
@@ -153,9 +193,10 @@ class DefaultSource extends RelationProvider with SchemaRelationProvider with Cr | |
| } | ||
|
|
||
| private def writeInSalesforceWave( | ||
| username: Option[String], | ||
| password: String, | ||
| authToken: Option[String], | ||
| serverUrl: String, | ||
| version: String, | ||
| datasetName: String, | ||
| appName: String, | ||
|
|
@@ -165,7 +206,7 @@ class DefaultSource extends RelationProvider with SchemaRelationProvider with Cr | |
| monitorJob: Boolean, | ||
| data: DataFrame, | ||
| metadata: Option[String]) { | ||
| val dataWriter = new DataWriter(username, password, authToken, serverUrl, version, datasetName, appName) | ||
|
|
||
| val metaDataJson = Utils.metadata(metadata, usersMetadataConfig, data.schema, datasetName) | ||
|
|
||
|
|
@@ -202,7 +243,7 @@ class DefaultSource extends RelationProvider with SchemaRelationProvider with Cr | |
|
|
||
| if (monitorJob) { | ||
| logger.info("Monitoring Job status in Salesforce wave") | ||
| if (Utils.monitorJob(writtenId.get, username, password, authToken, serverUrl, version)) { | ||
|
There was a problem hiding this comment. How is the authToken expiry should be handled?
Author
There was a problem hiding this comment. Great question, would you like automatic logic built into the spark layer or would you expect that to take place outside of spark? There was a problem hiding this comment. @wgorman I would like to handle in spark layer. There might be a chance that the access token got expired while fetching millions of rows |
||
| logger.info(s"Successfully dataset $datasetName has been processed in Salesforce Wave") | ||
| } else { | ||
| sys.error(s"Upload Job for dataset $datasetName failed in Salesforce Wave. Check Monitor Job in Salesforce Wave for more details") | ||
|
|
@@ -223,6 +264,17 @@ class DefaultSource extends RelationProvider with SchemaRelationProvider with Cr | |
| } | ||
| } | ||
|
|
||
|
|
||
| private def optionalParam(parameters: Map[String, String], envName: String, paramName: String) : Option[String] = { | ||
| val envProp = sys.env.get(envName) | ||
| if (envProp != null && envProp.isDefined) { | ||
| return Option(envProp.get) | ||
| } | ||
|
|
||
| val param = parameters.getOrElse(paramName, null) | ||
| return Option(param) | ||
| } | ||
|
|
||
| private def param(parameters: Map[String, String], envName: String, paramName: String) : String = { | ||
| val envProp = sys.env.get(envName) | ||
| if (envProp != null && envProp.isDefined) { | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We need to support existing login as well. i.e with username and password