Brug SparkR

SparkR er en R-pakke, der leverer en letvægts frontend til brug af Apache Spark fra R. SparkR tilbyder en distribueret dataframe-implementering, der understøtter operationer som udvælgelse, filtrering, aggregering og mere. SparkR understøtter også distribueret maskinlæring ved brug af MLlib.

Brug SparkR gennem Spark batchjob-definitioner eller med interaktive Microsoft Fabric-notebooks.

R-understøttelse er kun tilgængelig i Spark 3.1 eller nyere. R i Spark 2.4 understøttes ikke.

Forudsætninger

• Få et Microsoft Fabric abonnement. Eller tilmeld dig en gratis Microsoft Fabric prøveperiode.

• Log ind på Microsoft Fabric.

• Skift til Fabric ved at bruge experience-switcheren nederst til venstre på din startside.

  

• Åbn eller opret en notesbog. For at lære hvordan, se Sådan bruger du Microsoft Fabric notesbøger.

• Angiv sprogindstillingen til SparkR (R) for at ændre det primære sprog.

• Vedhæft din notesbog til et lakehouse. I venstre side skal du vælge Tilføj for at tilføje et eksisterende lakehouse eller for at oprette et lakehouse.

Læs og skriv SparkR DataFrames

Læs en SparkR-dataramme fra en lokal R-data.frame

Den nemmeste måde at oprette en DataFrame på er ved at konvertere en lokal R-data.frame til en Spark DataFrame.

# 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)

Læs og skriv SparkR DataFrame fra Lakehouse

Du kan gemme data på det lokale filsystem af klyngenoder. For at læse og skrive en SparkR DataFrame fra Lakehouse, brug read.df og-metoderne write.df . Disse metoder tager stien til den fil, der skal indlæses, og typen af datakilde. SparkR understøtter læsning af CSV-, JSON-, tekst- og parquetfiler oprindeligt.

Hvis du vil læse og skrive til et Lakehouse, skal du først føje det til din session. I venstre side af notesbogen skal du vælge Tilføj for at tilføje et eksisterende Lakehouse eller oprette et Lakehouse.

# 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 har tidyverse forudinstalleret. Du kan få adgang til Lakehouse-filer i dine velkendte R-pakker, såsom ved at læse og skrive Lakehouse-filer ved at bruge readr::read_csv() og 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)

Du kan også læse en SparkR Dataframe på din Lakehouse ved at bruge SparkSQL-forespørgsler.

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

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

head(waiting)

DataFrame-handlinger

SparkR DataFrames understøtter mange funktioner til struktureret databehandling. Her er nogle grundlæggende eksempler. Du kan finde en komplet liste i SparkR API-dokumentationerne.

Vælg rækker og kolonner

# 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))

Gruppering og sammenlægning

SparkR-datarammer understøtter mange almindeligt anvendte funktioner til at aggregere data efter gruppering. For eksempel kan du beregne et histogram af ventetiden i det trofaste datasæt, som vist i det følgende eksempel.

# 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)))

Kolonnehandlinger

SparkR tilbyder mange funktioner, som du kan anvende direkte på kolonner til databehandling og aggregering. I følgende eksempel vises brugen af grundlæggende aritmetiske funktioner.

# 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)

Anvend brugerdefineret funktion

SparkR understøtter flere typer brugerdefinerede funktioner:

Kør en funktion på et stort datasæt med dapply eller dapplyCollect

dapply

Anvend en funktion på hver partition i en SparkDataFrame. Funktionen, der skal anvendes på hver partition af bør SparkDataFrame kun have én parameter, som svarer til en data.frame for hver partition. Outputtet af funktionen skal være en data.frame. Skemaet angiver rækkeformatet for den resulterende SparkDataFrame. Den skal matche datatyperne for den returnerede værdi.

# 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

Ligesom dapply, anvend en funktion på hver partition af a SparkDataFrame og hent resultatet tilbage. Outputtet af funktionen skal være en data.frame. Men denne gang behøver du ikke at overlevere skemaet. Bemærk, at kan fejle, dapplyCollect hvis outputtene fra funktionen, der kører på alle partitionerne, ikke kan trækkes til driveren og passe i driverhukommelsen.

# 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)

Kør en funktion på en gruppering af store datasæt efter inputkolonne(er) med gapply eller gapplyCollect

gapply

Anvend en funktion på hver gruppe af en SparkDataFrame. Funktionen, der skal anvendes på hver gruppe af , SparkDataFrame bør kun have to parametre: grupperingsnøglen og et R data.frame , der svarer til denne nøgle. Du vælger grupperne fra SparkDataFrames kolonne(r). Outputtet af funktionen skal være en data.frame. Skemaet angiver rækkeformatet for den resulterende SparkDataFrame. Den skal repræsentere R-funktionens outputskema fra Spark-datatyper. Du sætter kolonnenavnene på det returnerede data.frame.

# 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

Ligesom gapply, anvend en funktion på hver gruppe i a SparkDataFrame og saml resultatet tilbage til et R data.frame. Outputtet af funktionen skal være en data.frame. Men du behøver ikke at sende skemaet. Bemærk, at kan fejle, gapplyCollect hvis outputtene fra funktionen, der kører på alle partitionerne, ikke kan trækkes til driveren og passe i driverhukommelsen.

# 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), ])

Kør lokale R-funktioner, der distribueres med spark.lapply

spark.lapply

Ligesom lapply i native R, spark.lapply kører en funktion over en liste af elementer og fordeler beregningerne ved hjælp af Spark. Den anvender en funktion på en måde, der ligner doParallel eller lapply ligner elementer i en liste. Resultaterne af alle beregningerne skal være i en enkelt maskine. Hvis den betingelse ikke er sand, kan du gøre noget som df <- createDataFrame(list) og så bruge dapply.

# 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)

Kør SQL-forespørgsler fra SparkR

Du kan registrere en SparkR DataFrame som en midlertidig visning, så du kan køre SQL-forespørgsler over dens data. Funktionen sql gør det muligt for applikationer at køre SQL-forespørgsler programmatisk og returnerer resultatet som en SparkR DataFrame.

# 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)

Maskinel indlæring

SparkR fremviser de fleste MLLib-algoritmer. Under hjelmen bruger SparkR MLlib til at oplære modellen.

I følgende eksempel kan du se, hvordan du opretter en Gaussisk GLM-model ved hjælp af SparkR. Hvis du vil køre lineær regression, skal du angive familie til "gaussian". Hvis du vil køre logistisk regression, skal du angive familie til "binomial". Når du bruger SparkML GLM , udfører SparkR automatisk one-hot kodning af kategoriske funktioner, så du ikke behøver at gøre det manuelt. Ud over String and Double-typer funktioner kan du også tilpasse MLlib Vector-funktioner for kompatibilitet med andre MLlib-komponenter.

For at lære mere om, hvilke maskinlæringsalgoritmer der understøttes, besøg dokumentationen for SparkR og 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)

Relateret indhold

• Sådan bruger du sparklyr
• Sådan bruger du Tidyverse
• Administration af R-bibliotek
• Opret R-visualisering
• Selvstudium: forudsigelse af avocadopris
• Selvstudium: forudsigelse af flyforsinkelse