How to implement Stanford CoreNLP wrapper for Apache Spark using sparklyr?

Question

I am trying to create a R package so I can use the Stanford CoreNLP wrapper for Apache Spark (by databricks) from R. I am using the sparklyr package to connect to my local Spark instance. I created a package with the following dependency function

spark_dependencies <- function(spark_version, scala_version, ...) {
  sparklyr::spark_dependency(
    jars = c(
      system.file(
        sprintf("stanford-corenlp-full/stanford-corenlp-3.6.0.jar"),
        package = "sparkNLP"
      ),
      system.file(
        sprintf("stanford-corenlp-full/stanford-corenlp-3.6.0-models.jar"),
        package = "sparkNLP"
      ),
      system.file(
        sprintf("stanford-corenlp-full/stanford-english-corenlp-2016-01-10-models.jar"),
        package = "sparkNLP"
      )
    ),
    packages = c(sprintf("databricks:spark-corenlp:0.2.0-s_%s", scala_version))
  )
}

In the logs I can see both databricks package an the dependent jars are loaded. I extracted all of coreNLP to the stanford-corenlp-full folder, so all depencies should be loaded correctly.

Ivy Default Cache set to: /Users/Bob/.ivy2/cache
The jars for the packages stored in: /Users/Bob/.ivy2/jars
:: loading settings :: url = jar:file:/Users/Bob/Library/Caches/spark/spark-2.0.0-bin-hadoop2.7/jars/ivy-2.4.0.jar!/org/apache/ivy/core/settings/ivysettings.xml
com.databricks#spark-csv_2.11 added as a dependency
com.amazonaws#aws-java-sdk-pom added as a dependency
databricks#spark-corenlp added as a dependency
:: resolving dependencies :: org.apache.spark#spark-submit-parent;1.0
    confs: [default]
    found com.databricks#spark-csv_2.11;1.3.0 in central
    found org.apache.commons#commons-csv;1.1 in central
    found com.univocity#univocity-parsers;1.5.1 in central
    found com.amazonaws#aws-java-sdk-pom;1.10.34 in central
    found databricks#spark-corenlp;0.2.0-s_2.11 in spark-packages
    found edu.stanford.nlp#stanford-corenlp;3.6.0 in central
    found com.io7m.xom#xom;1.2.10 in central
    found xml-apis#xml-apis;1.3.03 in central
    found xerces#xercesImpl;2.8.0 in central
    found xalan#xalan;2.7.0 in central
    found joda-time#joda-time;2.9 in central
    found de.jollyday#jollyday;0.4.7 in central
    found javax.xml.bind#jaxb-api;2.2.7 in central
    found com.googlecode.efficient-java-matrix-library#ejml;0.23 in central
    found javax.json#javax.json-api;1.0 in central
    found org.slf4j#slf4j-api;1.7.12 in central
    found com.google.protobuf#protobuf-java;2.6.1 in central
:: resolution report :: resolve 625ms :: artifacts dl 28ms
    :: modules in use:
    com.amazonaws#aws-java-sdk-pom;1.10.34 from central in [default]
    com.databricks#spark-csv_2.11;1.3.0 from central in [default]
    com.google.protobuf#protobuf-java;2.6.1 from central in [default]
    com.googlecode.efficient-java-matrix-library#ejml;0.23 from central in [default]
    com.io7m.xom#xom;1.2.10 from central in [default]
    com.univocity#univocity-parsers;1.5.1 from central in [default]
    databricks#spark-corenlp;0.2.0-s_2.11 from spark-packages in [default]
    de.jollyday#jollyday;0.4.7 from central in [default]
    edu.stanford.nlp#stanford-corenlp;3.6.0 from central in [default]
    javax.json#javax.json-api;1.0 from central in [default]
    javax.xml.bind#jaxb-api;2.2.7 from central in [default]
    joda-time#joda-time;2.9 from central in [default]
    org.apache.commons#commons-csv;1.1 from central in [default]
    org.slf4j#slf4j-api;1.7.12 from central in [default]
    xalan#xalan;2.7.0 from central in [default]
    xerces#xercesImpl;2.8.0 from central in [default]
    xml-apis#xml-apis;1.3.03 from central in [default]
    :: evicted modules:
    xml-apis#xml-apis;2.0.2 by [xml-apis#xml-apis;1.3.03] in [default]
    joda-time#joda-time;2.1 by [joda-time#joda-time;2.9] in [default]
    ---------------------------------------------------------------------
    |                  |            modules            ||   artifacts   |
    |       conf       | number| search|dwnlded|evicted|| number|dwnlded|
    ---------------------------------------------------------------------
    |      default     |   19  |   0   |   0   |   2   ||   16  |   0   |
    ---------------------------------------------------------------------
:: retrieving :: org.apache.spark#spark-submit-parent
    confs: [default]
    0 artifacts copied, 16 already retrieved (0kB/17ms)
16/10/16 00:08:15 INFO SparkContext: Running Spark version 2.0.0
16/10/16 00:08:15 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
16/10/16 00:08:15 INFO SecurityManager: Changing view acls to: Bob
16/10/16 00:08:15 INFO SecurityManager: Changing modify acls to: Bob
16/10/16 00:08:15 INFO SecurityManager: Changing view acls groups to: 
16/10/16 00:08:15 INFO SecurityManager: Changing modify acls groups to: 
16/10/16 00:08:15 INFO SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users  with view permissions: Set(Bob); groups with view permissions: Set(); users  with modify permissions: Set(Bob); groups with modify permissions: Set()
16/10/16 00:08:15 INFO Utils: Successfully started service 'sparkDriver' on port 54829.
16/10/16 00:08:15 INFO SparkEnv: Registering MapOutputTracker
16/10/16 00:08:15 INFO SparkEnv: Registering BlockManagerMaster
16/10/16 00:08:15 INFO DiskBlockManager: Created local directory at /private/var/folders/hs/yw76yd_95lscwclwg15n73tw0000gn/T/blockmgr-8df2c32b-19cb-4fdf-b321-85cb411b564a
16/10/16 00:08:15 INFO MemoryStore: MemoryStore started with capacity 366.3 MB
16/10/16 00:08:15 INFO SparkEnv: Registering OutputCommitCoordinator
16/10/16 00:08:16 INFO Utils: Successfully started service 'SparkUI' on port 4040.
16/10/16 00:08:16 INFO SparkUI: Bound SparkUI to 127.0.0.1, and started at http://127.0.0.1:4040
16/10/16 00:08:16 INFO SparkContext: Added JAR file:/Library/Frameworks/R.framework/Versions/3.3/Resources/library/sparkNLP/stanford-corenlp-full/stanford-corenlp-3.6.0.jar at spark://127.0.0.1:54829/jars/stanford-corenlp-3.6.0.jar with timestamp 1476569296302
16/10/16 00:08:16 INFO SparkContext: Added JAR file:/Library/Frameworks/R.framework/Versions/3.3/Resources/library/sparkNLP/stanford-corenlp-full/stanford-corenlp-3.6.0-models.jar at spark://127.0.0.1:54829/jars/stanford-corenlp-3.6.0-models.jar with timestamp 1476569296303
16/10/16 00:08:16 INFO SparkContext: Added JAR file:/Library/Frameworks/R.framework/Versions/3.3/Resources/library/sparkNLP/stanford-corenlp-full/stanford-english-corenlp-2016-01-10-models.jar at spark://127.0.0.1:54829/jars/stanford-english-corenlp-2016-01-10-models.jar with timestamp 1476569296303
16/10/16 00:08:16 INFO SparkContext: Added JAR file:/Users/Bob/.ivy2/jars/com.databricks_spark-csv_2.11-1.3.0.jar at spark://127.0.0.1:54829/jars/com.databricks_spark-csv_2.11-1.3.0.jar with timestamp 1476569296303
16/10/16 00:08:16 INFO SparkContext: Added JAR file:/Users/Bob/.ivy2/jars/databricks_spark-corenlp-0.2.0-s_2.11.jar at spark://127.0.0.1:54829/jars/databricks_spark-corenlp-0.2.0-s_2.11.jar with timestamp 1476569296304
16/10/16 00:08:16 INFO SparkContext: Added JAR file:/Users/Bob/.ivy2/jars/org.apache.commons_commons-csv-1.1.jar at spark://127.0.0.1:54829/jars/org.apache.commons_commons-csv-1.1.jar with timestamp 1476569296304
16/10/16 00:08:16 INFO SparkContext: Added JAR file:/Users/Bob/.ivy2/jars/com.univocity_univocity-parsers-1.5.1.jar at spark://127.0.0.1:54829/jars/com.univocity_univocity-parsers-1.5.1.jar with timestamp 1476569296304
16/10/16 00:08:16 INFO SparkContext: Added JAR file:/Users/Bob/.ivy2/jars/edu.stanford.nlp_stanford-corenlp-3.6.0.jar at spark://127.0.0.1:54829/jars/edu.stanford.nlp_stanford-corenlp-3.6.0.jar with timestamp 1476569296304
16/10/16 00:08:16 INFO SparkContext: Added JAR file:/Users/Bob/.ivy2/jars/com.google.protobuf_protobuf-java-2.6.1.jar at spark://127.0.0.1:54829/jars/com.google.protobuf_protobuf-java-2.6.1.jar with timestamp 1476569296304
16/10/16 00:08:16 INFO SparkContext: Added JAR file:/Users/Bob/.ivy2/jars/com.io7m.xom_xom-1.2.10.jar at spark://127.0.0.1:54829/jars/com.io7m.xom_xom-1.2.10.jar with timestamp 1476569296305
16/10/16 00:08:16 INFO SparkContext: Added JAR file:/Users/Bob/.ivy2/jars/joda-time_joda-time-2.9.jar at spark://127.0.0.1:54829/jars/joda-time_joda-time-2.9.jar with timestamp 1476569296305
16/10/16 00:08:16 INFO SparkContext: Added JAR file:/Users/Bob/.ivy2/jars/de.jollyday_jollyday-0.4.7.jar at spark://127.0.0.1:54829/jars/de.jollyday_jollyday-0.4.7.jar with timestamp 1476569296305
16/10/16 00:08:16 INFO SparkContext: Added JAR file:/Users/Bob/.ivy2/jars/com.googlecode.efficient-java-matrix-library_ejml-0.23.jar at spark://127.0.0.1:54829/jars/com.googlecode.efficient-java-matrix-library_ejml-0.23.jar with timestamp 1476569296305
16/10/16 00:08:16 INFO SparkContext: Added JAR file:/Users/Bob/.ivy2/jars/javax.json_javax.json-api-1.0.jar at spark://127.0.0.1:54829/jars/javax.json_javax.json-api-1.0.jar with timestamp 1476569296305
16/10/16 00:08:16 INFO SparkContext: Added JAR file:/Users/Bob/.ivy2/jars/org.slf4j_slf4j-api-1.7.12.jar at spark://127.0.0.1:54829/jars/org.slf4j_slf4j-api-1.7.12.jar with timestamp 1476569296306
16/10/16 00:08:16 INFO SparkContext: Added JAR file:/Users/Bob/.ivy2/jars/xml-apis_xml-apis-1.3.03.jar at spark://127.0.0.1:54829/jars/xml-apis_xml-apis-1.3.03.jar with timestamp 1476569296306
16/10/16 00:08:16 INFO SparkContext: Added JAR file:/Users/Bob/.ivy2/jars/xerces_xercesImpl-2.8.0.jar at spark://127.0.0.1:54829/jars/xerces_xercesImpl-2.8.0.jar with timestamp 1476569296306
16/10/16 00:08:16 INFO SparkContext: Added JAR file:/Users/Bob/.ivy2/jars/xalan_xalan-2.7.0.jar at spark://127.0.0.1:54829/jars/xalan_xalan-2.7.0.jar with timestamp 1476569296306
16/10/16 00:08:16 INFO SparkContext: Added JAR file:/Users/Bob/.ivy2/jars/javax.xml.bind_jaxb-api-2.2.7.jar at spark://127.0.0.1:54829/jars/javax.xml.bind_jaxb-api-2.2.7.jar with timestamp 1476569296306
16/10/16 00:08:16 INFO SparkContext: Added JAR file:/Library/Frameworks/R.framework/Versions/3.3/Resources/library/sparklyr/java/sparklyr-2.0-2.11.jar at spark://127.0.0.1:54829/jars/sparklyr-2.0-2.11.jar with timestamp 1476569296307
16/10/16 00:08:16 INFO Executor: Starting executor ID driver on host localhost
16/10/16 00:08:16 INFO Utils: Successfully started service 'org.apache.spark.network.netty.NettyBlockTransferService' on port 54830.
16/10/16 00:08:16 INFO NettyBlockTransferService: Server created on 127.0.0.1:54830
16/10/16 00:08:16 INFO BlockManagerMaster: Registering BlockManager BlockManagerId(driver, 127.0.0.1, 54830)
16/10/16 00:08:16 INFO BlockManagerMasterEndpoint: Registering block manager 127.0.0.1:54830 with 366.3 MB RAM, BlockManagerId(driver, 127.0.0.1, 54830)
16/10/16 00:08:16 INFO BlockManagerMaster: Registered BlockManager BlockManagerId(driver, 127.0.0.1, 54830)
16/10/16 00:08:16 WARN SparkContext: Use an existing SparkContext, some configuration may not take effect.
16/10/16 00:08:17 INFO HiveSharedState: Warehouse path is 'file:/Users/Bob/Documents/RPROJECTS/sparkNLP/spark-warehouse'

.

So I should be able to call the databricks sparkLib functions (located at com.databricks.spark.corenlp.functions )

However I can't seem to find the functions class when I call

library(sparkNLP) #the library I created
library(sparklyr)
sc <- spark_connect(master = "local")
invoke_new(sc,"com.databricks.spark.corenlp.functions")

I get an error message stating

Error: failed to invoke spark command
16/10/16 00:12:11 WARN cannot find matching constructor for class com.databricks.spark.corenlp.functions. Candidates are:
16/10/16 00:12:11 ERROR <init> on com.databricks.spark.corenlp.functions failed

I am not sure if the dependencies are not being loaded correctly or there is another problem.

Any help would be appreciated.

Below is my sessionInfo from RStudio

R version 3.3.1 (2016-06-21)
Platform: x86_64-apple-darwin13.4.0 (64-bit)
Running under: OS X 10.11.4 (El Capitan)

locale:
[1] en_US.UTF-8/en_US.UTF-8/en_US.UTF-8/C/en_US.UTF-8/en_US.UTF-8

attached base packages:
[1] stats     graphics  grDevices utils     datasets  methods   base     

other attached packages:
[1] sparklyr_0.4   sparkNLP_0.1.0

loaded via a namespace (and not attached):
 [1] Rcpp_0.12.7     digest_0.6.10   dplyr_0.5.0     withr_1.0.2     rprojroot_1.0-2
 [6] assertthat_0.1  rappdirs_0.3.1  R6_2.2.0        DBI_0.5-1       magrittr_1.5   
[11] config_0.2      tools_3.3.1     readr_1.0.0     yaml_2.1.13     parallel_3.3.1 
[16] tibble_1.2     

Answer 1

com.databricks.spark.corenlp.functions is an object, not a class, therefore calling is meaningless. This is basically what the error message says:

Error: java.lang.Exception: No matched constructor found for class com.databricks.spark.corenlp.functions

Instead you should access defined functions using invoke_static, for example:

invoke_static(sc,"com.databricks.spark.corenlp.functions", "cleanxml")

<jobj[15]>
org.apache.spark.sql.expressions.UserDefinedFunction
UserDefinedFunction(<function1>,StringType,Some(List(StringType)))

With example data borrowed from the official README

df <- copy_to(sc, tibble(
  id = 1,
  text = "<xml>Stanford University is located in California. It is a great university.</xml>"
))

you can define a wrapper like this:

sdf_cleanxml <- function(df, input_col, output_col) {
  sc <- df$src$con
  clean_xml <- invoke_static(sc,"com.databricks.spark.corenlp.functions", "cleanxml")
  arg <- list(invoke_static(sc, "org.apache.spark.sql.functions", "col", input_col))
  expr <- invoke(clean_xml, "apply", arg)
  df %>%
    spark_dataframe() %>% 
    invoke("withColumn", output_col, expr) %>%
    sdf_register()
}

and invoke it as follows:

sdf_cleanxml(df, "text", "text_clean")

# Source: spark<?> [?? x 3]
    id text                                 text_clean                         
  <dbl> <chr>                                <chr>                              
1     1 <xml>Stanford University is located… Stanford University is located in …

In practice though it might be simpler to just register required functions:

register_core_nlp <- function(sc) {
  funs <- c(
    "cleanxml", "tokenize", "ssplit", "pos", "lemma", "ner", "depparse",
    "coref", "natlog", "openie", "sentiment"
  )
  udf_registration <- sparklyr::invoke(sparklyr::spark_session(sc), "udf")
  for (fun in funs) {
    sparklyr::invoke(
      udf_registration, "register", fun,
      sparklyr::invoke_static(sc,"com.databricks.spark.corenlp.functions", fun)
    )
   }
}

register_core_nlp(sc)

and let SQL translation do the rest:

df %>% 
  transmute(doc = cleanxml(text)) %>%
  transmute(sen = explode(ssplit(doc))) %>%
  mutate(words = tokenize(sen), ner_tags = ner(sen), sentiment = sentiment(sen))

# Source: spark<?> [?? x 4]
  sen                                            words      ner_tags   sentiment
  <chr>                                          <list>     <list>         <int>
1 Stanford University is located in California . <list [7]> <list [7]>         1
2 It is a great university .                     <list [6]> <list [6]>         4