Merge pull request #13 from mlverse/updates

Updates

NAMESPACE

@@ -17,6 +17,7 @@ S3method(sample_frac,tbl_pyspark)
 S3method(sample_n,tbl_pyspark)
 S3method(sdf_copy_to,pyspark_connection)
 S3method(sdf_read_column,spark_pyjobj)
+S3method(sdf_register,spark_pyobj)
 S3method(spark_connect_method,spark_method_databricks_connect)
 S3method(spark_connect_method,spark_method_spark_connect)
 S3method(spark_connection,pyspark_connection)
@@ -99,6 +100,7 @@ importFrom(sparklyr,invoke_new)
 importFrom(sparklyr,random_string)
 importFrom(sparklyr,sdf_copy_to)
 importFrom(sparklyr,sdf_read_column)
+importFrom(sparklyr,sdf_register)
 importFrom(sparklyr,spark_connect_method)
 importFrom(sparklyr,spark_connection)
 importFrom(sparklyr,spark_dataframe)

R/install.R

@@ -28,7 +28,7 @@ install_pyspark <- function(envname = "r-sparklyr",
     "delta-spark"
   )
 
-  pip_options <- "--index-url https://packagemanager.posit.co/pypi/2023-06-01/simple"
+  pip_options <- "--index-url https://packagemanager.posit.co/pypi/2023-06-15/simple"
   # in cause user supplied pip_options in ...
   pip_options <- c(pip_options, list(...)$pip_options)
 

R/methods-dplyr.R

@@ -169,6 +169,18 @@ tbl_pyspark_sdf <- function(x) {
 
 tbl_temp_name <- function() glue("{temp_prefix()}{random_string()}")
 
+#' @importFrom sparklyr sdf_register
+#' @export
+sdf_register.spark_pyobj <- function(x, name = NULL) {
+  sc <- spark_connection(sc)
+  tbl_pyspark_temp(
+    x = x$pyspark_obj,
+    conn = sc,
+    tmp_name = name
+  )
+}
+
+
 tbl_pyspark_temp <- function(x, conn, tmp_name = NULL) {
   sc <- spark_connection(conn)
   if (is.null(tmp_name)) {

R/package.R

@@ -30,10 +30,8 @@
 NULL
 
 .onLoad <- function(...) {
-
   use_virtualenv("r-sparklyr", required = FALSE)
   use_condaenv("r-sparklyr", required = FALSE)
-
 }
 
 pysparklyr_env <- new.env()

R/spark-connect.R

@@ -81,7 +81,24 @@ py_spark_connect <- function(master,
       token = token,
       cluster_id = cluster_id
     )
-    python <- remote$getOrCreate()
+
+    check_rstudio <- try(RStudio.Version(), silent = TRUE)
+
+    if(inherits(check_rstudio, "try-error")) {
+      rstudio_chr <- NULL
+    } else {
+      rstudio_chr <- glue("rstudio/{check_rstudio$long_version}")
+    }
+
+    user_agent <- glue(
+      paste(
+        "sparklyr/{packageVersion('sparklyr')}",
+        rstudio_chr
+        )
+      )
+
+    agent <- remote$userAgent(user_agent)
+    python <- agent$getOrCreate()
     con_class <- "connect_databricks"
     master_label <- glue("Databricks Connect - Cluster: {cluster_id}")
   }

R/spark-pyobj.R

@@ -5,9 +5,11 @@ print.spark_pyobj <- function(x, ...) {
 
 #' @export
 sdf_read_column.spark_pyjobj <- function(x, column) {
-  sdf <- spark_dataframe(x)
-  col_df <- invoke(sdf, "select", column)
-  col_df <- collect(col_df)
+  col_df <- x %>%
+    spark_dataframe() %>%
+    invoke("select", column) %>%
+    collect()
+
   col_df[[column]]
 }