0001 #
0002 # Licensed to the Apache Software Foundation (ASF) under one or more
0003 # contributor license agreements. See the NOTICE file distributed with
0004 # this work for additional information regarding copyright ownership.
0005 # The ASF licenses this file to You under the Apache License, Version 2.0
0006 # (the "License"); you may not use this file except in compliance with
0007 # the License. You may obtain a copy of the License at
0008 #
0009 # http://www.apache.org/licenses/LICENSE-2.0
0010 #
0011 # Unless required by applicable law or agreed to in writing, software
0012 # distributed under the License is distributed on an "AS IS" BASIS,
0013 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
0014 # See the License for the specific language governing permissions and
0015 # limitations under the License.
0016 #
0017
0018 # S4 class representing Broadcast variables
0019
0020 # Hidden environment that holds values for broadcast variables
0021 # This will not be serialized / shipped by default
0022 .broadcastNames <- new.env()
0023 .broadcastValues <- new.env()
0024 .broadcastIdToName <- new.env()
0025
0026 # S4 class that represents a Broadcast variable
0027 #
0028 # Broadcast variables can be created using the broadcast
0029 # function from a \code{SparkContext}.
0030 #
0031 # @rdname broadcast-class
0032 # @seealso broadcast
0033 #
0034 # @param id Id of the backing Spark broadcast variable
0035 setClass("Broadcast", slots = list(id = "character"))
0036
0037 # @rdname broadcast-class
0038 # @param value Value of the broadcast variable
0039 # @param jBroadcastRef reference to the backing Java broadcast object
0040 # @param objName name of broadcasted object
0041 Broadcast <- function(id, value, jBroadcastRef, objName) {
0042 .broadcastValues[[id]] <- value
0043 .broadcastNames[[as.character(objName)]] <- jBroadcastRef
0044 .broadcastIdToName[[id]] <- as.character(objName)
0045 new("Broadcast", id = id)
0046 }
0047
0048 # @description
0049 # \code{value} can be used to get the value of a broadcast variable inside
0050 # a distributed function.
0051 #
0052 # @param bcast The broadcast variable to get
0053 # @rdname broadcast
0054 setMethod("value",
0055 signature(bcast = "Broadcast"),
0056 function(bcast) {
0057 if (exists(bcast@id, envir = .broadcastValues)) {
0058 get(bcast@id, envir = .broadcastValues)
0059 } else {
0060 NULL
0061 }
0062 })
0063
0064 # Internal function to set values of a broadcast variable.
0065 #
0066 # This function is used internally by Spark to set the value of a broadcast
0067 # variable on workers. Not intended for use outside the package.
0068 #
0069 # @rdname broadcast-internal
0070 # @seealso broadcast, value
0071
0072 # @param bcastId The id of broadcast variable to set
0073 # @param value The value to be set
0074 setBroadcastValue <- function(bcastId, value) {
0075 bcastIdStr <- as.character(bcastId)
0076 .broadcastValues[[bcastIdStr]] <- value
0077 }
0078
0079 # Helper function to clear the list of broadcast variables we know about
0080 # Should be called when the SparkR JVM backend is shutdown
0081 clearBroadcastVariables <- function() {
0082 bcasts <- ls(.broadcastNames)
0083 rm(list = bcasts, envir = .broadcastNames)
0084 }
