add HTML stripping and HTML entity unescaping custom UDF's and a model for adding UDF's internally in general.

fixed state handling to be index+statement-id specific so that multiple indexes could be managed separately from same statements
release 0.6.0-ALPHA

Author: apatrida

README.md

@@ -59,9 +59,7 @@ Lastly you specify `importSteps` which are queries that use any of the `sources`
           "username": "myusername",
           "password": "mypass"
         },
-        "driverJars": [
-          "/Users/myUserName/Downloads/mysql-connector-java-5.1.41/mysql-connector-java-5.1.41-bin.jar"
-        ],
+        "driverJars": [],  # MySQL and Postgres JARS are included automatically, this property can be omitted completely
         "tables": [
           {
             "sparkTable": "Users",
@@ -367,13 +365,21 @@ for full details on the supported SQL.
 A list of [SQL Functions](https://spark.apache.org/docs/2.1.0/api/java/org/apache/spark/sql/functions.html) 
 is available in raw API docs.  (_TODO: find better reference_)
 
+The Data Import Handler also defines some UDF functions for use within SQL:
+
+|Function|Description|
+|-------|-----------|
+|stripHtml(string)|Removes all HTML tags and returns only the text (including unescaping of HTML Entities)|
+|unescapeHtmlEntites(string)|Unescapes HTML entities found in the text|
+|fluffly(string)|A silly function that prepends the word "fluffly" to the text, used as a text function to mark values as being changed by processing|
+
 ### State Management and History:
 
 State for the `lastRun` value is per-statement and stored in the target Elasticsearch cluster for that statement.  An index
-will be created called `.kohesive-dih-state` which stores the last run state, a lock for current running statements, and
+will be created called `.kohesive-dih-state-v2` which stores the last run state, a lock for current running statements, and
 a log of all previous runs (success and failures, along with row count processed by the statement query).  
 
-You should inspect this log (index `.kohesive-dih-state` type `log`)if you wish to monitor the results of runs.
+You should inspect this log (index `.kohesive-dih-state-v2` type `log`)if you wish to monitor the results of runs.
 
 ### Parallelism
 

build.gradle

@@ -35,6 +35,8 @@ dependencies {
 
     compile group: 'org.apache.spark', name: "spark-sql_${version_scala}", version: '2.1.0'
 
+    compile group: 'org.jsoup', name: 'jsoup', version: version_jsoup
+
     resolutionRules group: 'com.netflix.nebula', name: 'gradle-resolution-rules', version: version_nebula_resolution
     resolutionRules files("${rootDir}/gradle-local-dependency-rules.json")
 

gradle.properties

@@ -14,6 +14,8 @@ version_elasticsearch=5.2.2
 version_okhttp=3.6.0
 version_scala=2.11
 
+version_jsoup=1.10.2
+
 version_slf4j=1.+
 version_logback=1.+
 version_junit=4.+

src/main/java/uy/kohesive/elasticsearch/dataimport/udf/Udfs.java

@@ -0,0 +1,22 @@
+package uy.kohesive.elasticsearch.dataimport.udf;
+
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.api.java.UDF1;
+import org.apache.spark.sql.api.java.UDF2;
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.DataTypes;
+import org.jsoup.Jsoup;
+import org.jsoup.nodes.Element;
+import org.jsoup.safety.Whitelist;
+
+public class Udfs {
+    public static <T extends String, RT extends String> void registerStringToStringUdf(final SparkSession spark, final String name, final UDF1<T, RT> f) {
+        spark.udf().register(name, f, DataTypes.StringType);
+    }
+
+    public static <T1 extends String, T2 extends String, RT extends String> void registerStringStringToStringUdf(final SparkSession spark, final String name, final UDF2<T1, T2, RT> f) {
+        spark.udf().register(name, f, DataTypes.StringType);
+    }
+
+
+}

src/main/kotlin/uy/kohesive/elasticsearch/dataimport/App.kt

@@ -10,6 +10,7 @@ import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.catalyst.parser.ParseException
 import org.elasticsearch.spark.sql.api.java.JavaEsSparkSQL
+import uy.kohesive.elasticsearch.dataimport.udf.Udfs
 import java.io.File
 import java.io.InputStream
 import java.io.InputStreamReader
@@ -71,9 +72,9 @@ class App {
 
         val lastRuns: Map<String, Instant> = cfg.importSteps.map { importStep ->
             importStep.statements.map { statement ->
-                val lastState = stateMap.get(statement.id)!!.readStateForStatement(uniqueId, statement.id)?.truncatedTo(ChronoUnit.SECONDS) ?: NOSTATE
+                val lastState = stateMap.get(statement.id)!!.readStateForStatement(uniqueId, statement)?.truncatedTo(ChronoUnit.SECONDS) ?: NOSTATE
                 println("  Statement ${statement.id} - ${statement.description}")
-                println("     LAST RUN: ${if (lastState === NOSTATE) "never" else lastState.toIsoString()}")
+                println("     LAST RUN: ${if (lastState == NOSTATE) "never" else lastState.toIsoString()}")
                 statement.id to lastState
             }
         }.flatten().toMap()
@@ -92,9 +93,7 @@ class App {
                     .master(sparkMaster).getOrCreate().use { spark ->
 
                 // add extra UDF functions
-                // TODO: udf's
-
-                // spark.udf().register("fluffy")
+                DataImportHandlerUdfs.registerSparkUdfs(spark)
 
                 // setup FILE inputs
                 println()
@@ -206,8 +205,13 @@ class App {
                         val sqlMinDate = Timestamp.from(lastRun).toString()
                         val sqlMaxDate = Timestamp.from(thisRunDate).toString()
 
-                        println("\n    Execute statement:  (range '$sqlMinDate' to '$sqlMaxDate')\n${statement.description.replaceIndent("        ")}")
-                        if (!stateMgr.lockStatement(uniqueId, statement.id)) {
+                        val dateMsg = if (lastRun == NOSTATE) {
+                           "range NEVER to '$sqlMaxDate'"
+                        } else {
+                           "range '$sqlMinDate' to '$sqlMaxDate'"
+                        }
+                        println("\n    Execute statement:  ($dateMsg)\n${statement.description.replaceIndent("        ")}")
+                        if (!stateMgr.lockStatement(uniqueId, statement)) {
                             System.err.println("        Cannot aquire lock for statement ${statement.id}")
                         } else {
                             try {
@@ -231,14 +235,15 @@ class App {
                                 val rowCount = sqlResults.count()
                                 println("        Rows processed: $rowCount")
 
-                                stateMgr.writeStateForStatement(uniqueId, statement.id, thisRunDate, "success", rowCount, null)
-                                stateMgr.logStatement(uniqueId, statement.id, thisRunDate, "sucess", rowCount, null)
+                                stateMgr.writeStateForStatement(uniqueId, statement, thisRunDate, "success", rowCount, null)
+                                stateMgr.logStatement(uniqueId, statement, thisRunDate, "sucess", rowCount, null)
                             } catch (ex: Exception) {
                                 val msg = ex.message ?: "unknown failure"
-                                stateMgr.writeStateForStatement(uniqueId, statement.id, thisRunDate, "error", 0, msg)
-                                stateMgr.logStatement(uniqueId, statement.id, thisRunDate, "error", 0, msg)
+                                stateMgr.writeStateForStatement(uniqueId, statement, lastRun, "error", 0, msg)
+                                stateMgr.logStatement(uniqueId, statement, thisRunDate, "error", 0, msg)
+                                System.err.println("\nProcess FAILED:  \n$msg\n")
                             } finally {
-                                stateMgr.unlockStatemnt(uniqueId, statement.id)
+                                stateMgr.unlockStatemnt(uniqueId, statement)
                             }
                         }
                     }

src/main/kotlin/uy/kohesive/elasticsearch/dataimport/State.kt

@@ -14,14 +14,16 @@ import java.time.Instant
 
 interface StateManager {
     fun init()
-    fun lockStatement(runId: String, statementId: String): Boolean
-    fun pingLockStatement(runId: String, statementId: String): Boolean
-    fun unlockStatemnt(runId: String, statementId: String)
-    fun writeStateForStatement(runId: String, statementId: String, lastRunStart: Instant, status: String, lastRowCount: Long, errMsg: String? = null)
-    fun readStateForStatement(runId: String, statementId: String): Instant?
-    fun logStatement(runId: String, statementId: String, lastRunStart: Instant, status: String, rowCount: Long, errMsg: String? = null)
+    fun lockStatement(runId: String, statement: EsImportStatement): Boolean
+    fun pingLockStatement(runId: String, statement: EsImportStatement): Boolean
+    fun unlockStatemnt(runId: String, statement: EsImportStatement)
+    fun writeStateForStatement(runId: String, statement: EsImportStatement, lastRunStart: Instant, status: String, lastRowCount: Long, errMsg: String? = null)
+    fun readStateForStatement(runId: String, statement: EsImportStatement): Instant?
+    fun logStatement(runId: String, statement: EsImportStatement, lastRunStart: Instant, status: String, rowCount: Long, errMsg: String? = null)
 }
 
+private fun EsImportStatement.stateKey(): String = this.indexName + "-" + this.id
+
 // TODO: better state management
 // This is NOT using the ES client because we do not want conflicts with Spark dependencies
 class ElasticSearchStateManager(val nodes: List<String>, val auth: AuthInfo?) : StateManager {
@@ -32,7 +34,7 @@ class ElasticSearchStateManager(val nodes: List<String>, val auth: AuthInfo?) :
         configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, true)
         configure(DeserializationFeature.READ_DATE_TIMESTAMPS_AS_NANOSECONDS, false)
     }
-    val STATE_INDEX = ".kohesive-dih-state"
+    val STATE_INDEX = ".kohesive-dih-state-v2"
 
     private fun OkHttpClient.get(url: String): Pair<Int, String> {
         val request = Request.Builder().url(url).build()
@@ -98,6 +100,7 @@ class ElasticSearchStateManager(val nodes: List<String>, val auth: AuthInfo?) :
                   "mappings": {
                      "state": {
                          "properties": {
+                             "targetIndex": { "type": "keyword" },
                              "statementId": { "type": "keyword" },
                              "lastRunDate": { "type": "date" },
                              "status": { "type": "keyword" },
@@ -108,6 +111,7 @@ class ElasticSearchStateManager(val nodes: List<String>, val auth: AuthInfo?) :
                      },
                      "log": {
                         "properties": {
+                             "targetIndex": { "type": "keyword" },
                              "statementId": { "type": "keyword" },
                              "runId": { "type": "keyword" },
                              "runDate": { "type": "date" },
@@ -118,6 +122,7 @@ class ElasticSearchStateManager(val nodes: List<String>, val auth: AuthInfo?) :
                      },
                      "lock": {
                         "properties": {
+                             "targetIndex": { "type": "keyword" },
                              "statementId": { "type": "keyword" },
                              "runId": { "type": "keyword" },
                              "lockDate": { "type": "date" }
@@ -137,7 +142,7 @@ class ElasticSearchStateManager(val nodes: List<String>, val auth: AuthInfo?) :
         ttlKillOldLocks()
     }
 
-    data class Lock(val runId: String, val statementId: String, val lockDate: Instant)
+    data class Lock(val runId: String, val targetIndex: String, val statementId: String, val lockDate: Instant)
 
     fun makeUrl(index: String, type: String) = "$url/$index/$type"
     fun makeUrl(type: String) = makeUrl(STATE_INDEX, type)
@@ -151,19 +156,19 @@ class ElasticSearchStateManager(val nodes: List<String>, val auth: AuthInfo?) :
         if (!delCode.isSuccess()) throw DataImportException("State manager failed, TTL delete query for locks failed\n$response")
     }
 
-    override fun lockStatement(runId: String, statementId: String): Boolean {
-        val lockUrl = "${makeUrl("lock")}/${statementId}"
-        val (code, response) = http.post("$lockUrl?op_type=create", JSON.writeValueAsString(Lock(runId, statementId, Instant.now())))
+    override fun lockStatement(runId: String, statement: EsImportStatement): Boolean {
+        val lockUrl = "${makeUrl("lock")}/${statement.stateKey()}"
+        val (code, response) = http.post("$lockUrl?op_type=create", JSON.writeValueAsString(Lock(runId, statement.indexName, statement.id, Instant.now())))
 
         if (!code.isSuccess()) {
             ttlKillOldLocks()
-            return pingLockStatement(runId, statementId)
+            return pingLockStatement(runId, statement)
         }
         return true
     }
 
-    override fun pingLockStatement(runId: String, statementId: String): Boolean {
-        val lockUrl = "${makeUrl("lock")}/${statementId}"
+    override fun pingLockStatement(runId: String, statement: EsImportStatement): Boolean {
+        val lockUrl = "${makeUrl("lock")}/${statement.stateKey()}"
         val (getCode, response) = http.get(lockUrl)
         if (getCode.isSuccess()) {
             val lock = mapFromSource<Lock>(response)
@@ -179,45 +184,47 @@ class ElasticSearchStateManager(val nodes: List<String>, val auth: AuthInfo?) :
                             "bool": {
                                "must": [
                                   { "term": { "runId": "$runId" } },
-                                  { "term": { "statementId": "$statementId" } }
+                                  { "term": { "targetIndex": "${statement.indexName}" } },
+                                  { "term": { "statementId": "${statement.id}" } }
                                ]
                             }
                         }
                     }
                 """)
                 if (!updCode.isSuccess()) {
-                    throw DataImportException("State manager failed, cannot acquire lock for $statementId - had conflict on pinging of lock\n$updResponse")
+                    throw DataImportException("State manager failed, cannot acquire lock for ${statement.stateKey()} - had conflict on pinging of lock\n$updResponse")
                 }
             } else {
-                throw DataImportException("State manager failed, cannot acquire lock for $statementId -- it is held by ${lock.runId} since ${lock.lockDate.toIsoString()}\n$response")
+                throw DataImportException("State manager failed, cannot acquire lock for ${statement.stateKey()} -- it is held by ${lock.runId} since ${lock.lockDate.toIsoString()}\n$response")
             }
         }
         return true
     }
 
-    override fun unlockStatemnt(runId: String, statementId: String) {
-        if (pingLockStatement(runId, statementId)) {
-            val lockUrl = "${makeUrl("lock")}/${statementId}?refresh"
+    override fun unlockStatemnt(runId: String, statement: EsImportStatement) {
+        if (pingLockStatement(runId, statement)) {
+            val lockUrl = "${makeUrl("lock")}/${statement.stateKey()}?refresh"
             val (code, response) = http.delete(lockUrl)
             if (!code.isSuccess()) {
-                throw DataImportException("State manager failed, cannot delete lock for $statementId\n$response")
+                throw DataImportException("State manager failed, cannot delete lock for ${statement.stateKey()}\n$response")
             }
         }
     }
 
-    data class State(val statementId: String, val lastRunDate: Instant, val status: String, val lastRunId: String, val lastErrorMesasge: String?, val lastRowCount: Long)
+    data class State(val targetIndex: String, val statementId: String, val lastRunDate: Instant, val status: String, val lastRunId: String, val lastErrorMesasge: String?, val lastRowCount: Long)
 
-    data class StateLog(val statementId: String, val runId: String, val runDate: Instant, val status: String, val errorMsg: String?, val rowCount: Long)
+    data class StateLog(val targetIndex: String, val statementId: String, val runId: String, val runDate: Instant, val status: String, val errorMsg: String?, val rowCount: Long)
 
-    override fun writeStateForStatement(runId: String, statementId: String, lastRunStart: Instant, status: String, lastRowCount: Long, errMsg: String?) {
-        val (code, response) = http.post("${makeUrl("state")}/${statementId}?refresh", JSON.writeValueAsString(State(statementId, lastRunStart, status, runId, errMsg, lastRowCount)))
+    override fun writeStateForStatement(runId: String, statement: EsImportStatement, lastRunStart: Instant, status: String, lastRowCount: Long, errMsg: String?) {
+        val (code, response) = http.post("${makeUrl("state")}/${statement.stateKey()}?refresh",
+                JSON.writeValueAsString(State(statement.indexName, statement.id, lastRunStart, status, runId, errMsg, lastRowCount)))
         if (!code.isSuccess()) {
-            throw DataImportException("State manager failed, cannot update state for $statementId\n$response")
+            throw DataImportException("State manager failed, cannot update state for ${statement.stateKey()}\n$response")
         }
     }
 
-    override fun readStateForStatement(runId: String, statementId: String): Instant? {
-        val (code, response) = http.get("${makeUrl("state")}/${statementId}")
+    override fun readStateForStatement(runId: String, statement: EsImportStatement): Instant? {
+        val (code, response) = http.get("${makeUrl("state")}/${statement.stateKey()}")
         if (code.isSuccess()) {
             val state = mapFromSource<State>(response)
             return state.lastRunDate
@@ -226,10 +233,11 @@ class ElasticSearchStateManager(val nodes: List<String>, val auth: AuthInfo?) :
         }
     }
 
-    override fun logStatement(runId: String, statementId: String, lastRunStart: Instant, status: String, rowCount: Long, errMsg: String?) {
-        val (code, response) = http.post("${makeUrl("log")}/${statementId}_run_${runId}?refresh", JSON.writeValueAsString(StateLog(statementId, runId, lastRunStart, status, errMsg, rowCount)))
+    override fun logStatement(runId: String, statement: EsImportStatement, lastRunStart: Instant, status: String, rowCount: Long, errMsg: String?) {
+        val (code, response) = http.post("${makeUrl("log")}/${statement.stateKey()}_run_${runId}?refresh",
+                JSON.writeValueAsString(StateLog(statement.indexName, statement.id, runId, lastRunStart, status, errMsg, rowCount)))
         if (!code.isSuccess()) {
-            throw DataImportException("State manager failed, cannot log state for $statementId\n$response")
+            throw DataImportException("State manager failed, cannot log state for ${statement.stateKey()}\n$response")
         }
     }
 }

src/main/kotlin/uy/kohesive/elasticsearch/dataimport/Udfs.kt

@@ -0,0 +1,38 @@
+package uy.kohesive.elasticsearch.dataimport
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.types.DataTypes
+import org.jsoup.Jsoup
+import org.jsoup.nodes.Document
+import org.jsoup.nodes.Entities
+import org.jsoup.parser.Parser
+import org.jsoup.safety.Whitelist
+import uy.kohesive.elasticsearch.dataimport.udf.Udfs
+
+object DataImportHandlerUdfs {
+    fun registerSparkUdfs(spark: SparkSession) {
+        Udfs.registerStringToStringUdf(spark, "fluffly", fluffly)
+        Udfs.registerStringToStringUdf(spark, "stripHtml", stripHtmlCompletely)
+        Udfs.registerStringToStringUdf(spark, "unescapeHtmlEntites", unescapeHtmlEntities)
+    }
+
+    @JvmStatic val whiteListMap = mapOf(
+            "none" to Whitelist.none(),
+            "basic" to Whitelist.basic(),
+            "basicwithimages" to Whitelist.basicWithImages(),
+            "relaxed" to Whitelist.relaxed(),
+            "simpletext" to Whitelist.simpleText(),
+            "simple" to Whitelist.simpleText()
+    )
+
+    @JvmStatic val fluffly = fun (v: String): String = "fluffly " + v
+
+    @JvmStatic val stripHtmlCompletely = fun (v: String): String {
+        return Jsoup.parseBodyFragment(v).text()
+    }
+
+    @JvmStatic val unescapeHtmlEntities = fun (v: String): String {
+        return Parser.unescapeEntities(v, false)
+    }
+
+}