Как применить функцию к каждой строке в SparkR?

Question

У меня есть файл в формате CSV, который содержит таблицу со столбцами «id», «timestamp», «action», «value» и «location». Я хочу, чтобы применить функцию к каждой строке таблицы, и я уже написал код в R следующим образом:

user <- read.csv(file_path,sep = ";") 
num <- nrow(user) 
curLocation <- "1" 
for(i in 1:num) { 
    row <- user[i,] 
    if(user$action != "power") 
     curLocation <- row$value 
    user[i,"location"] <- curLocation 
} 

Анализатор R скрипт работает отлично, и теперь я хочу, чтобы применить его SparkR. Однако я не мог получить доступ к i-й строке непосредственно в SparkR, и я не мог найти никакой функции для управления каждой строкой в ​​SparkR documentation.

Какой метод следует использовать для достижения того же эффекта, что и в сценарии R?

Кроме того, в соответствии с рекомендациями @chateaur, я пытался кодировать с помощью dapply функции следующим образом:

curLocation <- "1" 
schema <- structType(structField("Sequence","integer"), structField("ID","integer"), structField("Timestamp","timestamp"), structField("Action","string"), structField("Value","string"), structField("Location","string")) 
setLocation <- function(row, curLoc) { 
    if(row$Action != "power|battery|level"){ 
     curLoc <- row$Value 
    } 
    row$Location <- curLoc 
} 
bw <- dapply(user, function(row) { setLocation(row, curLocation)}, schema) 
head(bw) 

Тогда я получил ошибку:

я посмотрел предупреждающее сообщение условие имеет длину> 1, и будет использоваться только первый элемент и я нашел что-то https://stackoverflow.com/a/29969702/4942713. Это заставило меня задаться вопросом, является ли строка параметр в функции dapply представляют весь раздел моего фрейма данных вместо одного однорядного? Может быть, функция dapply не является желательным решением?

Позже я попытался изменить функцию, как рекомендовал @chateaur. Вместо использования dapply, я использовал dapplyCollect, который сэкономит мне силы на указание схемы. Оно работает!

changeLocation <- function(partitionnedDf) { 
    nrows <- nrow(partitionnedDf) 
    curLocation <- "1" 
    for(i in 1:nrows){ 
     row <- partitionnedDf[i,] 
     if(row$action != "power") { 
      curLocation <- row$value 
     } 
    partitionnedDf[i,"location"] <- curLocation 
    } 
    partitionnedDf 
} 

bw <- dapplyCollect(user, changeLocation) 

Answer 1

Scorpion775,

Вы должны разделить свой код sparkR. Не забывайте, что данные не обрабатываются одинаково в R и sparkR.

От: http://spark.apache.org/docs/latest/sparkr.html,

df <- read.df(csvPath, "csv", header = "true", inferSchema = "true", na.strings = "NA") 

Тогда вы можете посмотреть на dapply функции здесь: https://spark.apache.org/docs/2.1.0/api/R/dapply.html

Вот рабочий пример:

changeLocation <- function(partitionnedDf) { 
    nrows <- nrow(partitionnedDf) 
    curLocation <- as.integer(1) 

    # Loop over each row of the partitionned data frame 
    for(i in 1:nrows){ 
     row <- partitionnedDf[i,] 

     if(row[1] != "power") { 
      curLocation <- row[2] 
     } 
     partitionnedDf[i,3] <- curLocation 
    } 

    # Return modified data frame 
    partitionnedDf 
} 

# Load data 
df <- read.df("data.csv", "csv", header="false", inferSchema = "true") 

head(collect(df)) 

# Define schema of dataframe 
schema <- structType(structField("action", "string"), structField("value", "integer"), 
        structField("location", "integer")) 

# Change location of each row      
df2 <- dapply(df, changeLocation, schema) 

head(df2)