connection_shinyapp.R

rsApiUpdateDialog <- function(code) {
  if (exists(".rs.updateNewConnectionDialog")) {
    updateDialog <- get(".rs.updateNewConnectionDialog")
    updateDialog(code)
  }
}

rsApiShowDialog <- function(title, message, url = "") {
  if (exists(".rs.showDialog")) {
    showDialog <- get(".rs.showDialog")
    showDialog(title, message, url)
  }
}

rsApiShowPrompt <- function(title, message, default) {
  if (exists(".rs.showPrompt")) {
    showPrompt <- get(".rs.showPrompt")
    showPrompt(title, message, default)
  }
}

rsApiShowQuestion <- function(title, message, ok, cancel) {
  if (exists(".rs.showQuestion")) {
    showPrompt <- get(".rs.showQuestion")
    showPrompt(title, message, ok, cancel)
  }
}

rsApiReadPreference <- function(name, default) {
  value <- .rs.readUiPref(name)
  if (is.null(value)) default else value
}

rsApiWritePreference <- function(name, value) {
  if (!is.character(value)) {
    stop("Only character preferences are supported")
  }
  .rs.writeUiPref(name, value)
}

spark_ui_avaliable_versions <- function() {
  tryCatch({
    spark_available_versions()[,c("spark","hadoop")]
  }, error = function(e) {
    warning(e)
    spark_installed_versions()[,c("spark","hadoop")]
  })
}

spark_ui_spark_choices <- function() {
  availableVersions <- spark_ui_avaliable_versions()
  selected <- spark_default_version()[["spark"]]
  choiceValues <- unique(availableVersions[["spark"]])

  choiceNames <- choiceValues
  choiceNames <- lapply(
    choiceNames,
    function(e) if (e == selected) paste(e, "(Default)") else e
  )

  names(choiceValues) <- choiceNames

  choiceValues
}

spark_ui_hadoop_choices <- function(sparkVersion) {
  availableVersions <- spark_ui_avaliable_versions()

  selected <- spark_install_find(sparkVersion = sparkVersion, installedOnly = FALSE)$hadoopVersion

  choiceValues <- unique(availableVersions[availableVersions$spark == sparkVersion,][["hadoop"]])
  choiceNames <- choiceValues
  choiceNames <- lapply(
    choiceNames,
    function(e) if (length(selected) > 0 && e == selected) paste(e, "(Default)") else e
  )

  names(choiceValues) <- choiceNames

  choiceValues
}

spark_ui_default_connections <- function() {
  getOption(
    "sparklyr.ui.connections",
    getOption("rstudio.spark.connections")
  )
}

#' @import shiny
#' @import rstudioapi
connection_spark_ui <- function() {
  tags$div(
    tags$head(
      tags$style(
        HTML("
          body {
            background: none;

            font-family : \"Lucida Sans\", \"DejaVu Sans\", \"Lucida Grande\", \"Segoe UI\", Verdana, Helvetica, sans-serif;
            font-size : 12px;
            -ms-user-select : none;
            -moz-user-select : none;
            -webkit-user-select : none;
            user-select : none;

            margin: 0;
            margin-top: 5px;
          }

          .shiny-input-container {
            min-width: 100%;
            margin-bottom: 7px;
          }

          .shiny-input-container > .control-label {
            display: inline-block;
            width: 195px;
          }

          .shiny-input-container > div {
            display: inline-block;
            width: 300px;
          }

          #shiny-disconnected-overlay {
            display: none;
          }
        ")
      )
    ),
    selectInput(
      "master",
      "Master:",
      choices = c(
        list("local" = "local"),
        spark_ui_default_connections(),
        list("Cluster..." = "cluster")
      ),
      selectize = FALSE
    ),
    selectInput(
      "dbinterface",
      "DB Interface:",
      choices = c(
        "dplyr" = "dplyr",
        "(None)" = "none"
      ),
      selectize = FALSE,
      selected = rsApiReadPreference("sparklyr_dbinterface", "dplyr")
    ),
    div(
      style = "height: 10px"
    ),
    selectInput(
      "sparkversion",
      "Spark version:",
      choices = spark_ui_spark_choices(),
      selected = spark_default_version()$spark,
      selectize = FALSE
    ),
    selectInput(
      "hadoopversion",
      "Hadoop version:",
      choices = spark_ui_hadoop_choices(spark_default_version()$spark),
      selected = spark_default_version()$hadoop,
      selectize = FALSE
    )
  )
}

connection_spark_server <- function(input, output, session) {
  hasDefaultSparkVersion <- reactive({
    input$sparkversion == spark_default_version()$spark
  })

  hasDefaultHadoopVersion <- reactive({
    input$hadoopversion == spark_default_version()$hadoop
  })

  userInstallPreference <- NULL
  checkUserInstallPreference <- function(master, sparkSelection, hadoopSelection, prompt) {
    if (identical(master, "local") &&
        identical(versionInfo()$mode, "desktop") &&
        identical(spark_home(), NULL)) {

      installed <- spark_installed_versions()
      isInstalled <- nrow(installed[installed$spark == sparkSelection & installed$hadoop == hadoopSelection, ])

      if (!isInstalled) {
        if (prompt && identical(userInstallPreference, NULL)) {
          userInstallPreference <<- rsApiShowQuestion(
            "Install Spark Components",
            paste(
              "Spark ",
              sparkSelection,
              " for Hadoop ",
              hadoopSelection,
              " is not currently installed.",
              "\n\n",
              "Do you want to install this version of Spark?",
              sep = ""
            ),
            ok = "Install",
            cancel = "Cancel"
          )

          userInstallPreference
        }
        else if (identical(userInstallPreference, NULL)) {
          FALSE
        }
        else {
          userInstallPreference
        }
      }
      else {
        FALSE
      }
    }
    else {
      FALSE
    }
  }

  installSparkReactive <- reactive({
    sparkSelection <- input$sparkversion
    hadoopSelection <- input$hadoopversion
    master <- input$master

    checkUserInstallPreference(master, sparkSelection, hadoopSelection, FALSE)
  })

  generateCode <- function(master, dbInterface, sparkVersion, hadoopVersion, installSpark) {
    paste(
      "library(sparklyr)\n",
      if(dbInterface == "dplyr") "library(dplyr)\n" else "",
      if(installSpark)
        paste(
          "spark_install(version = \"",
          sparkVersion,
          "\", hadoop_version = \"",
          hadoopVersion,
          "\")\n",
          sep = ""
        )
      else "",
      "sc ",
      "<- ",
      "spark_connect(master = \"",
      master,
      "\"",
      if (!hasDefaultSparkVersion())
        paste(
          ", version = \"",
          sparkVersion,
          "\"",
          sep = ""
        )
      else "",
      if (!hasDefaultHadoopVersion())
        paste(
          ", hadoop_version = \"",
          hadoopVersion,
          "\"",
          sep = ""
        )
      else "",
      ")",
      sep = ""
    )
  }

  stateValuesReactive <- reactiveValues(codeInvalidated = 1)

  codeReactive <- reactive({
    master <- input$master
    dbInterface <- input$dbinterface
    sparkVersion <- input$sparkversion
    hadoopVersion <- input$hadoopversion
    installSpark <- installSparkReactive()
    codeInvalidated <- stateValuesReactive$codeInvalidated

    generateCode(master, dbInterface, sparkVersion, hadoopVersion, installSpark)
  })

  observe({
    rsApiUpdateDialog(codeReactive())
  })

  observe({
    if (identical(input$master, "cluster")) {
      if (identical(versionInfo()$mode, "desktop")) {
        rsApiShowDialog(
          "Connect to Spark",
          paste(
            "Connecting with a remote Spark cluster requires ",
            "an RStudio Server instance that is either within the cluster ",
            "or has a high bandwidth connection to the cluster.</p>",
            "<p>Please see the <strong>Using Spark with RStudio</strong> help ",
            "link below for additional details.</p>",
            sep = ""
          )
        )

        updateSelectInput(
          session,
          "master",
          selected = "local"
        )
      }
      else if (identical(spark_home(), NULL)) {
        rsApiShowDialog(
          "Connect to Spark",
          paste(
            "Connecting with a Spark cluster requires that you are on a system ",
            "able to communicate with the cluster in both directions, and ",
            "requires that the SPARK_HOME environment variable refers to a  ",
            "locally installed version of Spark that is configured to ",
            "communicate with the cluster.",
            "<p>Your system doesn't currently have the SPARK_HOME environment ",
            "variable defined. Please contact your system administrator to ",
            "ensure that the server is properly configured to connect with ",
            "the cluster.<p>",
            sep = ""
          )
        )

        updateSelectInput(
          session,
          "master",
          selected = "local"
        )
      }
      else {
        master <- rsApiShowPrompt(
          "Connect to Cluster",
          "Spark master:",
          "spark://local:7077"
        )

        updateSelectInput(
          session,
          "master",
          choices = masterChoices,
          selected = c(
            list(master = "master"),
            master,
            spark_ui_default_connections(),
            list("Cluster..." = "cluster")
          )
        )
      }
    }
  })

  currentSparkSelection <- NULL
  session$onFlushed(function() {
    if (!is_java_available()) {
      url <- ""

      message <- paste(
        "In order to connect to Spark ",
        "your system needs to have Java installed (",
        "no version of Java was detected).",
        sep = ""
      )

      if (identical(versionInfo()$mode, "desktop")) {
        message <- paste(
          message,
          "<p>Please contact your server administrator to request the ",
          "installation of Java on this system.</p>",
          sep = "")

        url <- java_install_url()
      } else {
        message <- paste(
          message,
          "<p>Please contact your server administrator to request the ",
            "installation of Java on this system.</p>",
          sep = "")
      }

      rsApiShowDialog(
        "Java Required for Spark Connections",
        message,
        url
      )
    }

    currentSparkSelection <<- spark_default_version()$spark
  })

  observe({
    # Scope this reactive to only changes to spark version
    sparkVersion <- input$sparkversion
    master <- input$master

    # Don't change anything while initializing
    if (!identical(currentSparkSelection, NULL)) {
      currentSparkSelection <<- sparkVersion

      hadoopDefault <- spark_install_find(sparkVersion = currentSparkSelection, installedOnly = FALSE)$hadoopVersion

      updateSelectInput(
        session,
        "hadoopversion",
        choices = spark_ui_hadoop_choices(currentSparkSelection),
        selected = hadoopDefault
      )

      stateValuesReactive$codeInvalidated <<- isolate({
        stateValuesReactive$codeInvalidated + 1
      })

      checkUserInstallPreference(master, currentSparkSelection, spark_default_version()$hadoop, TRUE)
    }
  })

  observe({
    # Scope this reactive to only changes to hadoop version, by using a ref
    # to currentSparkSelection we force this code to be always executed
    # after the spark version changes.
    hadoopVersion <- input$hadoopversion
    master <- input$master

    # Don't change anything while initializing
    if (!identical(currentSparkSelection, NULL)) {
      checkUserInstallPreference(master, currentSparkSelection, hadoopVersion, TRUE)

      stateValuesReactive$codeInvalidated <<- isolate({
        stateValuesReactive$codeInvalidated + 1
      })
    }
  })

  observe({
    rsApiWritePreference("sparklyr_dbinterface", input$dbinterface)
  })
}

#' A Shiny app that can be used to construct a \code{spark_connect} statement
#'
#' @export
#'
#' @import shiny
#'
#' @keywords internal
connection_spark_shinyapp <- function() {
  shinyApp(connection_spark_ui, connection_spark_server)
}