Migrowanie z platformy SparkR do interfejsu sparklyr

SparkR został opracowany w ramach platformy Apache Spark, a jego projekt jest znany użytkownikom języka Scala i Python, ale potencjalnie mniej intuicyjny dla praktyków języka R. Ponadto usługa SparkR jest przestarzała na platformie Spark 4.0.

Z kolei sparklyr koncentruje się na zapewnieniu bardziej przyjaznego dla języka R środowiska. dplyr Wykorzystuje składnię, która jest znana użytkownikom tidyverse z wzorcami takimi jak select(), filter() i mutate() w przypadku operacji na ramkach danych.

Sparklyr to zalecany pakiet języka R do pracy z platformą Apache Spark. Na tej stronie wyjaśniono różnice między SparkR a sparklyr w ramach API Spark oraz przedstawiono informacje o migracji kodu.

Konfigurowanie środowiska

Installation

Jeśli jesteś w obszarze roboczym usługi Azure Databricks, instalacja nie jest wymagana. Załaduj sparklyr za pomocą library(sparklyr). Aby zainstalować pakiet sparklyr lokalnie poza Azure Databricks, możesz zobaczyć Wprowadzenie.

Nawiązywanie połączenia z platformą Spark

Połącz się z platformą Spark za pomocą interfejsu sparklyr w obszarze roboczym usługi Databricks lub lokalnie przy użyciu usługi Databricks Connect:

Obszar roboczy:

library(sparklyr)
sc <- spark_connect(method = "databricks")

Databricks Connect:

sc <- spark_connect(method = "databricks_connect")

Aby uzyskać więcej informacji i rozszerzony samouczek dotyczący usługi Databricks Connect z interfejsem sparklyr, zobacz Wprowadzenie.

Odczytywanie i zapisywanie danych

sparklyr ma rodzinę funkcji spark_read_*() i spark_write_*() do ładowania i zapisywania danych, w przeciwieństwie do ogólnych funkcji read.df() i write.df() SparkR. Istnieją również unikatowe funkcje do tworzenia ramek danych platformy Spark lub tymczasowych widoków spark SQL z ramek danych języka R w pamięci.

Zadanie	SparkR	sparklyr
Kopiowanie danych na platformę Spark	createDataFrame()	copy_to()
Tworzenie widoku tymczasowego	createOrReplaceTempView()	Użyj metody invoke() bezpośrednio
Zapisywanie danych w tabeli	saveAsTable()	spark_write_table()
Zapisywanie danych w określonym formacie	write.df()	spark_write_<format>()
Odczytywanie danych z tabeli	tableToDF()	tbl() lub spark_read_table()
Odczytywanie danych z określonego formatu	read.df()	spark_read_<format>()

Ładowanie danych

Aby przekonwertować ramkę danych języka R na ramkę danych platformy Spark lub utworzyć tymczasowy widok poza ramką danych w celu zastosowania do niej kodu SQL:

SparkR

mtcars_df <- createDataFrame(mtcars)

sparklyr

mtcars_tbl <- copy_to(
  sc,
  df = mtcars,
  name = "mtcars_tmp",
  overwrite = TRUE,
  memory = FALSE
)

copy_to() Tworzy widok tymczasowy przy użyciu określonej nazwy. Możesz użyć nazwy do odwołowania się do danych, jeśli używasz języka SQL bezpośrednio (na przykład sdf_sql()). Ponadto copy_to() buforuje dane, ustawiając parametr memory na TRUE.

Tworzenie widoków

W poniższych przykładach kodu pokazano sposób tworzenia widoków tymczasowych:

SparkR

createOrReplaceTempView(mtcars_df, "mtcars_tmp_view")

sparklyr

spark_dataframe(mtcars_tbl) |>
  invoke("createOrReplaceTempView", "mtcars_tmp_view")

Zapisywanie danych

W poniższych przykładach kodu pokazano, jak są zapisywane dane:

SparkR

# Save a DataFrame to Unity Catalog
saveAsTable(
  mtcars_df,
  tableName = "<catalog>.<schema>.<table>",
  mode = "overwrite"
)

# Save a DataFrame to local filesystem using Delta format
write.df(
  mtcars_df,
  path = "file:/<path/to/save/delta/mtcars>",
  source = "delta",
  mode = "overwrite"
)

sparklyr

# Save tbl_spark to Unity Catalog
spark_write_table(
  mtcars_tbl,
  name = "<catalog>.<schema>.<table>",
  mode = "overwrite"
)

# Save tbl_spark to local filesystem using Delta format
spark_write_delta(
  mtcars_tbl,
  path = "file:/<path/to/save/delta/mtcars>",
  mode = "overwrite"
)

# Use DBI
library(DBI)
dbWriteTable(
  sc,
  value = mtcars_tbl,
  name = "<catalog>.<schema>.<table>",
  overwrite = TRUE
)

Odczytywanie danych

W poniższych przykładach kodu pokazano, jak dane są odczytywane:

SparkR

# Load a Unity Catalog table as a DataFrame
tableToDF("<catalog>.<schema>.<table>")

# Load csv file into a DataFrame
read.df(
  path = "file:/<path/to/read/csv/data.csv>",
  source = "csv",
  header = TRUE,
  inferSchema = TRUE
)

# Load Delta from local filesystem as a DataFrame
read.df(
  path = "file:/<path/to/read/delta/mtcars>",
  source = "delta"
)

# Load data from a table using SQL - Databricks recommendeds using `tableToDF`
sql("SELECT * FROM <catalog>.<schema>.<table>")

sparklyr

# Load table from Unity Catalog with dplyr
tbl(sc, "<catalog>.<schema>.<table>")

# or using `in_catalog`
tbl(sc, in_catalog("<catalog>", "<schema>", "<table>"))

# Load csv from local filesystem as tbl_spark
spark_read_csv(
  sc,
  name = "mtcars_csv",
  path = "file:/<path/to/csv/mtcars>",
  header = TRUE,
  infer_schema = TRUE
)

# Load delta from local filesystem as tbl_spark
spark_read_delta(
  sc,
  name = "mtcars_delta",
  path = "file:/tmp/test/sparklyr1"
)

# Load data using SQL
sdf_sql(sc, "SELECT * FROM <catalog>.<schema>.<table>")

Przetwarzanie danych

Wybieranie i filtrowanie

SparkR

# Select specific columns
select(mtcars_df, "mpg", "cyl", "hp")

# Filter rows where mpg > 20
filter(mtcars_df, mtcars_df$mpg > 20)

sparklyr

# Select specific columns
mtcars_tbl |>
  select(mpg, cyl, hp)

# Filter rows where mpg > 20
mtcars_tbl |>
  filter(mpg > 20)

Dodawanie kolumn

SparkR

# Add a new column 'power_to_weight' (hp divided by wt)
withColumn(mtcars_df, "power_to_weight", mtcars_df$hp / mtcars_df$wt)

sparklyr

# Add a new column 'power_to_weight' (hp divided by wt)
mtcars_tbl |>
  mutate(power_to_weight = hp / wt)

Grupowanie i agregacja

SparkR

# Calculate average mpg and hp by number of cylinders
mtcars_df |>
  groupBy("cyl") |>
  summarize(
    avg_mpg = avg(mtcars_df$mpg),
    avg_hp = avg(mtcars_df$hp)
  )

sparklyr

# Calculate average mpg and hp by number of cylinders
mtcars_tbl |>
  group_by(cyl) |>
  summarize(
    avg_mpg = mean(mpg),
    avg_hp = mean(hp)
  )

Dołącza

Załóżmy, że mamy inny zestaw danych z etykietami cylindrów, które chcemy dołączyć do mtcars.

SparkR

# Create another DataFrame with cylinder labels
cylinders <- data.frame(
  cyl = c(4, 6, 8),
  cyl_label = c("Four", "Six", "Eight")
)
cylinders_df <- createDataFrame(cylinders)

# Join mtcars_df with cylinders_df
join(
  x = mtcars_df,
  y = cylinders_df,
  mtcars_df$cyl == cylinders_df$cyl,
  joinType = "inner"
)

sparklyr

# Create another SparkDataFrame with cylinder labels
cylinders <- data.frame(
  cyl = c(4, 6, 8),
  cyl_label = c("Four", "Six", "Eight")
)
cylinders_tbl <- copy_to(sc, cylinders, "cylinders", overwrite = TRUE)

# join mtcars_df with cylinders_tbl
mtcars_tbl |>
  inner_join(cylinders_tbl, by = join_by(cyl))

Funkcje zdefiniowane przez użytkownika (UDF)

Aby utworzyć funkcję niestandardową na potrzeby kategoryzacji:

# Define the custom function
categorize_hp <- function(df)
  df$hp_category <- ifelse(df$hp > 150, "High", "Low") # a real-world example would use case_when() with mutate()
  df

SparkR

Usługa SparkR wymaga jawnego zdefiniowania schematu wyjściowego przed zastosowaniem funkcji:

# Define the schema for the output DataFrame
schema <- structType(
  structField("mpg", "double"),
  structField("cyl", "double"),
  structField("disp", "double"),
  structField("hp", "double"),
  structField("drat", "double"),
  structField("wt", "double"),
  structField("qsec", "double"),
  structField("vs", "double"),
  structField("am", "double"),
  structField("gear", "double"),
  structField("carb", "double"),
  structField("hp_category", "string")
)

# Apply the function across partitions
dapply(
  mtcars_df,
  func = categorize_hp,
  schema = schema
)

# Apply the same function to each group of a DataFrame. Note that the schema is still required.
gapply(
  mtcars_df,
  cols = "hp",
  func = categorize_hp,
  schema = schema
)

sparklyr

# Load Arrow to avoid cryptic errors
library(arrow)

# Apply the function over data.
# By default this applies to each partition.
mtcars_tbl |>
  spark_apply(f = categorize_hp)

# Apply the function over data
# Use `group_by` to apply data over groups
mtcars_tbl |>
  spark_apply(
    f = summary,
    group_by = "hp" # This isn't changing the resulting output as the functions behavior is applied to rows independently.
  )

spark.lapply() vs spark_apply()

W SparkR spark.lapply() działa na listach języka R zamiast na ramkach danych. Nie ma bezpośredniego odpowiednika w interfejsie sparklyr, ale możesz osiągnąć podobne zachowanie, pracując z spark_apply() ramką danych, która zawiera unikatowe identyfikatory i grupowanie według tych identyfikatorów. W niektórych przypadkach operacje oparte na wierszach mogą również zapewniać porównywalną funkcjonalność. Aby uzyskać więcej informacji na temat spark_apply(), zobacz Dystrybucja obliczeń języka R.

SparkR

# Define a list of integers
numbers <- list(1, 2, 3, 4, 5)

# Define a function to apply
square <- function(x)
  x * x

# Apply the function over list using Spark
spark.lapply(numbers, square)

sparklyr

# Create a DataFrame of given length
sdf <- sdf_len(sc, 5, repartition = 1)

# Apply function to each partition of the DataFrame
# spark_apply() defaults to processing data based on number of partitions.
# In this case it will return a single row due to repartition = 1.
spark_apply(sdf, f = nrow)

# Apply function to each row (option 1)
# To force behaviour like spark.lapply() you can create a DataFrame with N rows and force grouping with group_by set to a unique row identifier. In this case it's the id column automatically generated by sdf_len()). This will return N rows.
spark_apply(sdf, f = nrow, group_by = "id")

# Apply function to each row (option 2)
# This requires writing a function that operates across rows of a data.frame, in some occasions this may be faster relative to option 1. Specifying group_by in optional for this example. This example does not require rowwise(), but is just to illustrate one method to force computations to be for every row.
row_func <- function(df)
  df |>
    dplyr::rowwise() |>
    dplyr::mutate(x = id * 2)

spark_apply(sdf, f = row_func)

Uczenie maszynowe

Pełne przykłady SparkR i sparklyr dla uczenia maszynowego znajdują się w Przewodniku Spark ML i dokumentacji sparklyr.

Uwaga / Notatka

Jeśli nie używasz MLlib Sparka, Databricks zaleca użycie UDF do trenowania przy użyciu wybranej przez siebie biblioteki (na przykład xgboost).

Regresja liniowa

SparkR

# Select features
training_df <- select(mtcars_df, "mpg", "hp", "wt")

# Fit the model using Generalized Linear Model (GLM)
linear_model <- spark.glm(training_df, mpg ~ hp + wt, family = "gaussian")

# View model summary
summary(linear_model)

sparklyr

# Select features
training_tbl <- mtcars_tbl |>
  select(mpg, hp, wt)

# Fit the model using Generalized Linear Model (GLM)
linear_model <- training_tbl |>
  ml_linear_regression(response = "mpg", features = c("hp", "wt"))

# View model summary
summary(linear_model)

klastrowanie metodą k-średnich

SparkR

# Apply KMeans clustering with 3 clusters using mpg and hp as features
kmeans_model <- spark.kmeans(mtcars_df, mpg ~ hp, k = 3)

# Get cluster predictions
predict(kmeans_model, mtcars_df)

sparklyr

# Use mpg and hp as features
features_tbl <- mtcars_tbl |>
  select(mpg, hp)

# Assemble features into a vector column
features_vector_tbl <- features_tbl |>
  ft_vector_assembler(
    input_cols = c("mpg", "hp"),
    output_col = "features"
  )

# Apply K-Means clustering
kmeans_model <- features_vector_tbl |>
  ml_kmeans(features_col = "features", k = 3)

# Get cluster predictions
ml_predict(kmeans_model, features_vector_tbl)

Wydajność i optymalizacja

Zbieranie

Zarówno sparkR, jak i sparklyr służą collect() do konwertowania ramek danych platformy Spark na ramki danych języka R. Zbieraj z powrotem tylko małe ilości danych do ramek danych języka R, inaczej sterownik Spark'owi zabraknie pamięci.

Aby zapobiec błędom braku pamięci, usługa SparkR ma wbudowane optymalizacje w środowisku Databricks Runtime, które ułatwiają zbieranie danych lub wykonywanie funkcji zdefiniowanych przez użytkownika.

Aby zapewnić optymalną wydajność za pomocą pakietu sparklyr do zbierania danych i funkcji zdefiniowanych przez użytkownika w środowisku Databricks Runtime w wersjach poniżej 14.3 LTS, załaduj arrow pakiet:

library(arrow)

Partycjonowanie w pamięci

SparkR

# Repartition the SparkDataFrame based on 'cyl' column
repartition(mtcars_df, col = mtcars_df$cyl)

# Repartition the SparkDataFrame to number of partitions
repartition(mtcars_df, numPartitions = 10)

# Coalesce the DataFrame to number of partitions
coalesce(mtcars_df, numPartitions = 1)

# Get number of partitions
getNumPartitions(mtcars_df)

sparklyr

# Repartition the tbl_spark based on 'cyl' column
sdf_repartition(mtcars_tbl, partition_by = "cyl")

# Repartition the tbl_spark to number of partitions
sdf_repartition(mtcars_tbl, partitions = 10)

# Coalesce the tbl_spark to number of partitions
sdf_coalesce(mtcars_tbl, partitions = 1)

# Get number of partitions
sdf_num_partitions(mtcars_tbl)

Caching

SparkR

# Cache the DataFrame in memory
cache(mtcars_df)

sparklyr

# Cache the tbl_spark in memory
tbl_cache(sc, name = "mtcars_tmp")