Verwenden von SparkR

SparkR ist ein R-Paket, das ein leichtes Frontend für die Verwendung von Apache Spark von R bietet. SparkR bietet eine verteilte Datenframeimplementierung, die Vorgänge wie Auswahl, Filterung, Aggregation und mehr unterstützt. SparkR unterstützt auch verteiltes maschinelles Lernen mithilfe von MLlib.

Verwenden Sie SparkR über Spark Batch-Auftragsdefinitionen oder mit interaktiven Microsoft Fabric-Notizbüchern.

R-Unterstützung ist nur in Spark 3.1 oder höher verfügbar. R in Spark 2.4 wird nicht unterstützt.

Voraussetzungen

• Rufen Sie ein Microsoft Fabric-Abonnement ab. Oder registrieren Sie sich für eine kostenlose Microsoft Fabric Testversion.

• Melden Sie sich bei Microsoft Fabric an.

• Wechseln Sie zu Fabric, indem Sie den Benutzeroberflächenschalter auf der unteren linken Seite Ihrer Startseite verwenden.

  

• Öffnen oder erstellen Sie ein Notebook. Informationen dazu finden Sie unter Verwendung von Microsoft Fabric-Notebooks.

• Legen Sie zum Ändern der primären Sprache die Sprachoption auf SparkR (R) fest.

• Verbinden Sie Ihr Notebook mit einem Lakehouse. Wählen Sie auf der linken Seite Hinzufügen aus, um ein vorhandenes Lakehouse hinzuzufügen oder ein Lakehouse zu erstellen.

Lesen und Schreiben von SparkR-Dataframes

Ein SparkR-DataFrame aus einem lokalen R-Datenrahmen lesen.

Die einfachste Möglichkeit zum Erstellen eines Dataframe besteht in der Konvertierung eines lokalen R-Dataframe (data.frame) in SparkDataFrame.

# load SparkR package
library(SparkR)

# read a SparkR DataFrame from a local R data.frame
df <- createDataFrame(faithful)

# displays the content of the DataFrame
display(df)

Lesen und Schreiben von SparkR DataFrame aus Lakehouse

Sie können Daten im lokalen Dateisystem von Clusterknoten speichern. Um einen SparkR DataFrame aus Lakehouse zu lesen und zu schreiben, verwenden Sie die read.df und write.df methoden. Diese Methoden verwenden den Pfad der zu ladenden Datei und den Typ der Datenquelle. SparkR unterstützt nativ das Lesen von CSV-, JSON-, Text- und Parquet-Dateien.

Um in einem Lakehouse zu lesen und zu schreiben, fügen Sie das Lakehouse zuerst zu Ihrer Sitzung hinzu. Wählen Sie auf der linken Seite des Notebooks die Option Hinzufügen aus, um ein vorhandenes Lakehouse hinzuzufügen oder ein neues zu erstellen.

# write data in CSV using relative path for Spark
temp_csv_spark<-"Files/data/faithful.csv"
write.df(df, temp_csv_spark ,source="csv", mode = "overwrite", header = "true")

# read data in CSV using relative path for Spark
faithfulDF_csv <- read.df(temp_csv_spark, source= "csv", header = "true", inferSchema = "true")

# displays the content of the DataFrame
display(faithfulDF_csv)

# write data in parquet using an ABFS path
temp_parquet_spark<-"abfss://xxx/xxx/data/faithful.parquet"
write.df(df, temp_parquet_spark ,source="parquet", mode = "overwrite", header = "true")

# read data in parquet using an ABFS path
faithfulDF_pq <- read.df(temp_parquet_spark, source= "parquet", header = "true", inferSchema = "true")

# displays the content of the DataFrame
display(faithfulDF_pq)

Microsoft Fabric ist tidyverse vorinstalliert. Sie können auf Lakehouse-Dateien in Ihren vertrauten R-Paketen zugreifen, z. B. durch Lesen und Schreiben von Lakehouse-Dateien mithilfe readr::read_csv() und readr::write_csv().

# read data in CSV using API path
# To find the path, navigate to the csv file, right click, and  Copy File API path.
temp_csv_api<-'/lakehouse/default/Files/data/faithful.csv/part-00000-d8e09a34-bd63-41bd-8cf8-f4ed2ef90e6c-c000.csv'
faithfulDF_API <- readr::read_csv(temp_csv_api)

# display the content of the R data.frame
head(faithfulDF_API)

Sie können auch einen SparkR Dataframe auf Ihrem Lakehouse lesen, indem Sie SparkSQL-Abfragen verwenden.

# Register earlier df as temp view
createOrReplaceTempView(df, "eruptions")

# Create a df using a SparkSQL query
waiting <- sql("SELECT * FROM eruptions")

head(waiting)

DataFrame-Vorgänge

Spark-DataFrames unterstützen viele Funktionen für die strukturierte Datenverarbeitung. Es folgen einige einfache Beispiele. Eine vollständige Liste finden Sie in der API-Dokumentation für SparkR.

Auswählen von Zeilen und Spalten

# Select only the "waiting" column
head(select(df,df$waiting))

# Pass in column name as strings
head(select(df, "waiting"))

# Filter to only retain rows with waiting times longer than 70 mins
head(filter(df, df$waiting > 70))

Gruppierung und Aggregation

SparkR-Dataframes unterstützen viele gängige Funktionen zum Aggregieren von Daten nach der Gruppierung. Sie können z. B. ein Histogramm der Wartezeit im treuen Dataset berechnen, wie im folgenden Beispiel gezeigt.

# we use the `n` operator to count the number of times each waiting time appears
head(summarize(groupBy(df, df$waiting), count = n(df$waiting)))

# we can also sort the output from the aggregation to get the most common waiting times
waiting_counts <- summarize(groupBy(df, df$waiting), count = n(df$waiting))
head(arrange(waiting_counts, desc(waiting_counts$count)))

Spaltenvorgänge (Column operations)

SparkR bietet viele Funktionen, die Sie direkt auf Spalten für die Datenverarbeitung und -aggregation anwenden können. Das folgende Beispiel zeigt die Verwendung grundlegender arithmetischer Funktionen.

# convert waiting time from hours to seconds.
# you can assign this to a new column in the same DataFrame
df$waiting_secs <- df$waiting * 60
head(df)

Anwenden einer benutzerdefinierten Funktion

SparkR unterstützt verschiedene Arten von benutzerdefinierten Funktionen:

Ausführen einer Funktion für ein großes Dataset mit dapply oder dapplyCollect

dapply

Wenden Sie eine Funktion auf jede Partition eines SparkDataFrame an. Die Funktion, die auf jede Partition von SparkDataFrame angewendet wird, sollte nur einen Parameter haben, der einem data.frame für die jeweilige Partition entspricht. Die Ausgabe der Funktion sollte ein data.frame sein. Das Schema gibt das Zeilenformat des resultierenden SparkDataFrame an. Sie muss mit den Datentypen des zurückgegebenen Werts übereinstimmen.

# convert waiting time from hours to seconds
df <- createDataFrame(faithful)
schema <- structType(structField("eruptions", "double"), structField("waiting", "double"),
                     structField("waiting_secs", "double"))

# apply UDF to DataFrame
df1 <- dapply(df, function(x) { x <- cbind(x, x$waiting * 60) }, schema)
head(collect(df1))

dapplyCollect

Wenden Sie wie bei dapply eine Funktion auf jede Partition eines SparkDataFrame an und führen Sie die Ergebnisse wieder zusammen. Die Ausgabe der Funktion sollte ein data.frame sein. Dieses Mal müssen Sie das Schema jedoch nicht übergeben. Beachten Sie, dass dapplyCollect fehlschlagen kann, wenn die Ausgaben der Funktion, die auf allen Partitionen ausgeführt wird, nicht auf den Treiber übertragen werden können und nicht in den Arbeitsspeicher des Treibers passen.

# convert waiting time from hours to seconds
# apply UDF to DataFrame and return a R's data.frame
ldf <- dapplyCollect(
         df,
         function(x) {
           x <- cbind(x, "waiting_secs" = x$waiting * 60)
         })
head(ldf, 3)

Ausführen einer Funktion für ein großes Dataset mit Gruppierung nach Eingabespalte(n) mit gapply oder gapplyCollect

gapply

Wenden Sie eine Funktion auf jede SparkDataFrame-Gruppe an. Die Funktion, die auf jede Gruppe des SparkDataFrame angewendet wird, sollte nur zwei Parameter haben: den Gruppierungsschlüssel und ein R data.frame, das zu diesem Schlüssel gehört. Sie wählen die Gruppen aus SparkDataFrames Spalte(n) aus. Die Ausgabe der Funktion sollte ein data.frame sein. Das Schema gibt das Zeilenformat des resultierenden SparkDataFrame an. Es muss das Ausgabeschema der R-Funktion aus Spark-Datentypen darstellen. Sie legen die Spaltennamen für das zurückgegebene data.frame fest.

# determine six waiting times with the largest eruption time in minutes.
schema <- structType(structField("waiting", "double"), structField("max_eruption", "double"))
result <- gapply(
    df,
    "waiting",
    function(key, x) {
        y <- data.frame(key, max(x$eruptions))
    },
    schema)
head(collect(arrange(result, "max_eruption", decreasing = TRUE)))

gapplyCollect

Wenden Sie wie bei gapply eine Funktion auf jede Gruppe eines SparkDataFrame an und führen Sie die Ergebnisse wieder in einem R-data.frame zusammen. Die Ausgabe der Funktion sollte ein data.frame sein. Sie müssen das Schema jedoch nicht übergeben. Beachten Sie, dass gapplyCollect fehlschlagen kann, wenn die Ausgaben der Funktion, die auf allen Partitionen ausgeführt wird, nicht auf den Treiber übertragen werden können und nicht in den Arbeitsspeicher des Treibers passen.

# determine six waiting times with the largest eruption time in minutes.
result <- gapplyCollect(
    df,
    "waiting",
    function(key, x) {
        y <- data.frame(key, max(x$eruptions))
        colnames(y) <- c("waiting", "max_eruption")
        y
    })
head(result[order(result$max_eruption, decreasing = TRUE), ])

Ausführen lokal verteilter R-Funktionen mit „spark.lapply“

spark.lapply

Ähnlich wie lapply in nativem R führt spark.lapply eine Funktion auf einer Liste von Elementen aus und verteilt die Berechnungen mithilfe von Spark. Sie wendet eine Funktion auf Elemente einer Liste in ähnlicher Weise wie doParallel oder lapply an. Alle Berechnungsergebnisse sollten auf einem einzelnen Computer ausgeführt werden können. Wenn diese Bedingung nicht zutrifft, können Sie so etwas wie df <- createDataFrame(list) tun und dann dapply verwenden.

# perform distributed training of multiple models with spark.lapply. Here, we pass
# a read-only list of arguments which specifies family the generalized linear model should be.
families <- c("gaussian", "poisson")
train <- function(family) {
  model <- glm(Sepal.Length ~ Sepal.Width + Species, iris, family = family)
  summary(model)
}
# return a list of model's summaries
model.summaries <- spark.lapply(families, train)

# print the summary of each model
print(model.summaries)

Ausführen von SQL-Abfragen über SparkR

Sie können einen SparkR DataFrame als temporäre Ansicht registrieren, sodass Sie SQL-Abfragen über die zugehörigen Daten ausführen können. Mit der sql Funktion können Anwendungen SQL-Abfragen programmgesteuert ausführen und das Ergebnis als SparkR DataFrame zurückgeben.

# Register earlier df as temp view
createOrReplaceTempView(df, "eruptions")

# Create a df using a SparkSQL query
waiting <- sql("SELECT waiting FROM eruptions where waiting>70 ")

head(waiting)

Maschinelles Lernen

SparkR macht die meisten MLLib-Algorithmen verfügbar. Im Hintergrund verwendet SparkR die MLlib, um das Modell zu trainieren.

Das folgende Beispiel zeigt, wie Sie mit SparkR ein gaußsches generalisiertes lineares Modell (GLM) erstellen. Legen Sie zum Ausführen einer linearen Regression „family“ auf "gaussian" fest. Legen Sie zum Ausführen einer logistischen Regression „family“ auf "binomial" fest. Wenn Sie SparkML GLM SparkR verwenden, führt SparkR automatisch eine One-Hot-Codierung kategorialer Merkmale durch, sodass Sie dies nicht manuell tun müssen. Neben Features vom Typ „String“ und „Double“ können Sie das Modell zur Kompatibilität mit anderen MLlib-Komponenten auch auf MLlib-Vektor-Features anpassen.

Weitere Informationen dazu, welche Machine Learning-Algorithmen unterstützt werden, finden Sie in der Dokumentation für SparkR und MLlib.

# create the DataFrame
cars <- cbind(model = rownames(mtcars), mtcars)
carsDF <- createDataFrame(cars)

# fit a linear model over the dataset.
model <- spark.glm(carsDF, mpg ~ wt + cyl, family = "gaussian")

# model coefficients are returned in a similar format to R's native glm().
summary(model)

Zugehöriger Inhalt

• Verwenden von Sparklyr
• Verwenden von Tidyverse
• R-Bibliotheksverwaltung
• Erstellen von R-Visualisierungen
• Tutorial: Avocado-Preisvorhersage
• Tutorial: Flugverspätungsvorhersage