![\"SparkR](\"https:\/\/data-flair.training\/blogs\/wp-content\/uploads\/sites\/2\/2018\/01\/Spark-DataFrame-Operations-in-SparkR-01-1.jpg\")
<\/a>SparkR DataFrame operations<\/p><\/div>\n
You must test your Spark Learning so far\u00a0<\/a><\/strong><\/p>\n Data is organized as a distributed collection of data into named columns. That we call on SparkDataFrame. Basically, it is as same as a table in a relational database or a data frame in R<\/a><\/strong>. Moreover, we can construct a DataFrame<\/strong><\/a> from a wide array of sources. For example structured data files, tables in Hive, external databases. Also, existing local R data frames are used for construction<\/span><\/p>\n Basically, for structured data processing, SparkDataFrames supports many functions.<\/span> \u00a0Let\u2019s discuss some basic examples of it:<\/p>\n # Create the SparkDataFrame Have a look at Spark Paired RDD<\/a><\/strong><\/p>\n To aggregate data after grouping, SparkR DataFrame support various commonly used functions. <\/span> Do you know about Spark Executor<\/a><\/strong><\/p>\n There are some Functions that can directly apply to columns. It works for data processing and during aggregation. Here, an example shows the use of basic arithmetic functions.<\/span> There are various kinds of User-Defined Functions supported \u00a0in SparkR:<\/span> Basically, Dapply applies a function to each partition of a SparkDataFrame.<\/span> That function to be applied to each partition of the SparkDataFrame. There is one parameter required, to which a data.frame corresponds to each partition will be passed. The output of function should be a data.frame. The schema specifies the row format of the resulting a SparkDataFrame. It must match to data types of returned value.<\/p>\n Have a look at SparkContext As same as dapply, apply a function to each partition of a SparkDataFrame. Also collect the result back. The output of function should be a data.frame. However, Schema is not required to be passed. <\/span> Do you know about Spark Machine Learning with R b. Run a given function on a large dataset grouping by input column(s) and using gapply or gapplyCollect<\/strong><\/p>\n Apply a function to each group of a SparkDataFrame. The function is to be applied to each group of the SparkDataFrame and should have only two parameters: grouping key and R data.frame corresponding to that key. The groups are chosen from SparkDataFrames column(s). The output of function should be a data.frame. The schema specifies the row format of the resulting SparkDataFrame. It must represent R function\u2019s output schema on the basis of Spark data types. The column names of the returned data.frame are set by the user.<\/span><\/p>\n Best preparation for upcoming Spark Interview As same as gapply, it applies a function to each partition of a SparkDataFrame. \u00a0Also collects the result back to R data.frame. Here, also the output of the function should be a data.frame. Although, the schema is not required here to be passed. <\/span> Let’s revise the Spark Stage c.\u00a0 Run local R functions distributed using spark.lapply<\/strong> You must know about Spark GraphX API We can also register a SparkR DataFrame as a temporary view in Spark SQL<\/strong><\/a>. Basically, that allows us to run SQL queries over its data. Moreover, to run SQL queries programmatically, sql function enables applications. Also, returns the result as a SparkDataFrame.<\/span> Let’s check the comparison of Spark\u00a0Batch Processing and Real-time Processing<\/a><\/strong><\/p>\n So, this was all in SparkR DataFrame Tutorial. Hope you like our explanation.<\/p>\n As a result, we have seen all the SparkR DataFrame Operations. Also, we have seen several examples to understand the topic well. Although, if any query occurs, feel free to ask in the comment section.<\/p>\n See also –\u00a0<\/strong><\/p>\n2. SparkR DataFrame<\/h2>\n
3. SparkR DataFrame Operations<\/h2>\n
i. Selecting rows, columns<\/h3>\n
\ndf <- as.DataFrame(faithful)
\n# Get basic information about the SparkDataFrame
\ndf
\n## SparkDataFrame[eruptions:double, waiting:double]
\n# Select only the “eruptions” column
\nhead(select(df, df$eruptions))
\n## \u00a0eruptions
\n##1 \u00a0\u00a0\u00a0\u00a03.600
\n##2 \u00a0\u00a0\u00a0\u00a01.800
\n##3 \u00a0\u00a0\u00a0\u00a03.333
\n# You can also pass in column name as strings
\nhead(select(df, “eruptions”))
\n# Filter the SparkDataFrame to only retain rows with wait times shorter than 50 mins
\nhead(filter(df, df$waiting < 50))
\n## \u00a0eruptions waiting
\n##1 \u00a0\u00a0\u00a0\u00a01.750 \u00a0\u00a0\u00a0\u00a0\u00a047
\n##2 \u00a0\u00a0\u00a0\u00a01.750 \u00a0\u00a0\u00a0\u00a0\u00a047
\n##3 \u00a0\u00a0\u00a0\u00a01.867 \u00a0\u00a0\u00a0\u00a0\u00a048<\/p>\nii. Grouping, Aggregation<\/h3>\n
\nFor example, as shown below we are computing a histogram of the waiting time in the faithful dataset.<\/span>
\n<\/span>
\n<\/span># We use the `n` operator to count the number of times each waiting time appears
\nhead(summarize(groupBy(df, df$waiting), count = n(df$waiting)))
\n## \u00a0waiting count
\n##1 \u00a0\u00a0\u00a0\u00a0\u00a070 \u00a0\u00a0\u00a0\u00a04
\n##2 \u00a0\u00a0\u00a0\u00a0\u00a067 \u00a0\u00a0\u00a0\u00a01
\n##3 \u00a0\u00a0\u00a0\u00a0\u00a069 \u00a0\u00a0\u00a0\u00a02
\n# We can also sort the output from the aggregation to get the most common waiting times
\nwaiting_counts <- summarize(groupBy(df, df$waiting), count = n(df$waiting))
\nhead(arrange(waiting_counts, desc(waiting_counts$count)))
\n## \u00a0\u00a0waiting count
\n##1 \u00a0\u00a0\u00a0\u00a0\u00a078 \u00a0\u00a0\u00a015
\n##2 \u00a0\u00a0\u00a0\u00a0\u00a083 \u00a0\u00a0\u00a014
\n##3 \u00a0\u00a0\u00a0\u00a0\u00a081 \u00a0\u00a0\u00a013<\/p>\niii. Operating on Columns<\/h3>\n
\n<\/span>
\n<\/span># Convert waiting time from hours to seconds.
\n# Note that we can assign this to a new column in the same SparkDataFrame
\ndf$waiting_secs <- df$waiting * 60
\nhead(df)
\n## \u00a0eruptions waiting waiting_secs
\n##1 \u00a0\u00a0\u00a0\u00a03.600 \u00a0\u00a0\u00a0\u00a0\u00a079 \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a04740
\n##2 \u00a0\u00a0\u00a0\u00a01.800 \u00a0\u00a0\u00a0\u00a0\u00a054 \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a03240
\n##3 \u00a0\u00a0\u00a0\u00a03.333 \u00a0\u00a0\u00a0\u00a0\u00a074 \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a04440<\/p>\niv. Applying User-Defined Function<\/h3>\n
\n<\/span>
\n<\/span>a. Run a given function on a large dataset using dapply or dapplyCollect<\/strong><\/p>\n\n
\n<\/a><\/strong># Convert waiting time from hours to seconds.
\n# Note that we can apply UDF to DataFrame.
\nschema <- structType(structField(“eruptions”, “double”), structField(“waiting”, “double”),
\nstructField(“waiting_secs”, “double”))
\ndf1 <- dapply(df, function(x) { x <- cbind(x, x$waiting * 60) }, schema)
\nhead(collect(df1))
\n## \u00a0eruptions waiting waiting_secs
\n##1 \u00a0\u00a0\u00a0\u00a03.600 \u00a0\u00a0\u00a0\u00a0\u00a079 \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a04740
\n##2 \u00a0\u00a0\u00a0\u00a01.800 \u00a0\u00a0\u00a0\u00a0\u00a054 \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a03240
\n##3 \u00a0\u00a0\u00a0\u00a03.333 \u00a0\u00a0\u00a0\u00a0\u00a074 \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a04440
\n##4 \u00a0\u00a0\u00a0\u00a02.283 \u00a0\u00a0\u00a0\u00a0\u00a062 \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a03720
\n##5 \u00a0\u00a0\u00a0\u00a04.533 \u00a0\u00a0\u00a0\u00a0\u00a085 \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a05100
\n##6 \u00a0\u00a0\u00a0\u00a02.883 \u00a0\u00a0\u00a0\u00a0\u00a055 \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a03300<\/p>\n\n
\nNote: \u00a0dapplyCollect can fail if the output of UDF run on all the partition. It cannot be pulled to the driver and fit in driver memory.<\/span><\/p>\n
\n<\/a><\/strong>
\n<\/span># Convert waiting time from hours to seconds.
\n# Note that we can apply UDF to DataFrame and return a R’s data.frame
\nldf <- dapplyCollect(
\ndf,
\nfunction(x) {
\nx <- cbind(x, “waiting_secs” = x$waiting * 60)
\n})
\nhead(ldf, 3)
\n## \u00a0eruptions waiting waiting_secs
\n##1 \u00a0\u00a0\u00a0\u00a03.600 \u00a0\u00a0\u00a0\u00a0\u00a079 \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a04740
\n##2 \u00a0\u00a0\u00a0\u00a01.800 \u00a0\u00a0\u00a0\u00a0\u00a054 \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a03240
\n##3 \u00a0\u00a0\u00a0\u00a03.333 \u00a0\u00a0\u00a0\u00a0\u00a074 \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a04440<\/p>\n\n
\n<\/a><\/strong>
\n<\/span># Determine six waiting times with the largest eruption time in minutes.
\nschema <- structType(structField(“waiting”, “double”), structField(“max_eruption”, “double”))
\nresult <- gapply(
\ndf,
\n“waiting”,
\nfunction(key, x) {
\ny <- data.frame(key, max(x$eruptions))
\n},
\nschema)
\nhead(collect(arrange(result, “max_eruption”, decreasing = TRUE)))
\n## \u00a0\u00a0\u00a0waiting \u00a0\u00a0max_eruption
\n##1 \u00a0\u00a0\u00a0\u00a0\u00a064 \u00a0\u00a0\u00a0\u00a0\u00a0\u00a05.100
\n##2 \u00a0\u00a0\u00a0\u00a0\u00a069 \u00a0\u00a0\u00a0\u00a0\u00a0\u00a05.067
\n##3 \u00a0\u00a0\u00a0\u00a0\u00a071 \u00a0\u00a0\u00a0\u00a0\u00a0\u00a05.033
\n##4 \u00a0\u00a0\u00a0\u00a0\u00a087 \u00a0\u00a0\u00a0\u00a0\u00a0\u00a05.000
\n##5 \u00a0\u00a0\u00a0\u00a0\u00a063 \u00a0\u00a0\u00a0\u00a0\u00a0\u00a04.933
\n##6 \u00a0\u00a0\u00a0\u00a0\u00a089 \u00a0\u00a0\u00a0\u00a0\u00a0\u00a04.900<\/p>\n\n
\nNote:<\/b> gapplyCollect can fail if the output of UDF run on all the partition. It cannot be pulled to the driver. Also, fit in driver memory.<\/span><\/p>\n
\n<\/a><\/strong>
\n<\/span># Determine six waiting times with the largest eruption time in minutes.
\nresult <- gapplyCollect(
\ndf,
\n“waiting”,
\nfunction(key, x) {
\ny <- data.frame(key, max(x$eruptions))
\ncolnames(y) <- c(“waiting”, “max_eruption”)
\ny
\n})
\nhead(result[order(result$max_eruption, decreasing = TRUE), ])
\n## \u00a0\u00a0\u00a0waiting \u00a0\u00a0max_eruption
\n##1 \u00a0\u00a0\u00a0\u00a0\u00a064 \u00a0\u00a0\u00a0\u00a0\u00a0\u00a05.100
\n##2 \u00a0\u00a0\u00a0\u00a0\u00a069 \u00a0\u00a0\u00a0\u00a0\u00a0\u00a05.067
\n##3 \u00a0\u00a0\u00a0\u00a0\u00a071 \u00a0\u00a0\u00a0\u00a0\u00a0\u00a05.033
\n##4 \u00a0\u00a0\u00a0\u00a0\u00a087 \u00a0\u00a0\u00a0\u00a0\u00a0\u00a05.000
\n##5 \u00a0\u00a0\u00a0\u00a0\u00a063 \u00a0\u00a0\u00a0\u00a0\u00a0\u00a04.933
\n##6 \u00a0\u00a0\u00a0\u00a0\u00a089 \u00a0\u00a0\u00a0\u00a0\u00a0\u00a04.900<\/p>\n
\nSpark.lapply<\/strong>
\nAs Similar as lapply in native R, spark.lapply runs a function over a list of elements. Also distributes the computations with Spark. The manner in which it Applies a function is similar to doParallel or lapply to elements of a list. \u00a0Basically, all the results of all computations should fit on a single machine.<\/span><\/p>\n
\n<\/a><\/strong>
\n<\/span># Perform distributed training of multiple models with spark.lapply. Here, we pass
\n# a read-only list of arguments which specifies family the generalized linear model should be.
\nfamilies <- c(“gaussian”, “poisson”)
\ntrain <- function(family) {
\nmodel <- glm(Sepal.Length ~ Sepal.Width + Species, iris, family = family)
\nsummary(model)
\n}
\n# Return a list of model’s summaries
\nmodel.summaries <- spark.lapply(families, train)
\n# Print the summary of each model
\nprint(model.summaries)<\/p>\nv. Running SQL Queries from SparkR<\/h3>\n
\n<\/span>
\n<\/span># Load a JSON file
\npeople <- read.df(“.\/examples\/src\/main\/resources\/people.json”, “json”)
\n# Register this SparkDataFrame as a temporary view.
\ncreateOrReplaceTempView(people, “people”)
\n# SQL statements can be run by using the sql method
\nteenagers <- sql(“SELECT name FROM people WHERE age >= 13 AND age <= 19”)
\nhead(teenagers)
\n## \u00a0\u00a0\u00a0name
\n##1 Justin<\/p>\n4. Conclusion – SparkR DataFrame<\/h2>\n