kenthzhang

args <- commandArgs(trailingOnly = TRUE)
api <- as.integer(args[1]) # 1-rxExecBy, 2-gapply
size <- as.integer(args[2]) # size of data set, million rows
algo <- as.integer(args[3]) # 1-rxSummary, 2-rxLogit, 3-mean
key <- as.integer(args[4]) # 1-DayOfWeek, 2-Dest, 3-Origin+Dest

keys <- list(c("DayOfWeek"), c("Dest"), c("Origin", "Dest"))
apis <- list("rxExecBy", "gapply")
algos <- list("rxSummary", "rxLogit", "mean")
formulas <- list(ArrDelay15 ~ UniqueCarrier + Origin + Dest, ArrDelay15 ~ UniqueCarrier + Origin + DayOfWeek)

kenthzhang

cc <- rxSparkConnect(reset = TRUE, nameNode="wasb://benchmark@airlinecranrscaler.blob.core.windows.net/", driverMem="4g", numExecutors=4, executorCores=5, executorMem="16g", executorOverheadMem="4g")

# prepare hive data 1M, 2M, 5M, 10M, 100M
hdfs <- RxHdfsFileSystem(hostName="wasb://benchmark@airlinecranrscaler.blob.core.windows.net/")

airText <- RxTextData("wasb://benchmark@airlinecranrscaler.blob.core.windows.net/airline/csv/airOT1M", firstRowIsColNames=T, fileSystem=hdfs)
airHive <- RxHiveData(table="airOT1M")
rxDataStep(airText, airHive)

airText <- RxTextData("wasb://benchmark@airlinecranrscaler.blob.core.windows.net/airline/csv/airOT2M", firstRowIsColNames=T, fileSystem=hdfs)

kenthzhang

cc <- rxSparkConnect(reset = TRUE)
hdfsFileSystem <- RxHdfsFileSystem()
colInfo <- list(
    DayOfWeek = list(
        type = "factor",
        levels = c(
            "Monday",
            "Tuesday",
            "Wednesday",
            "Thursday",

kenthzhang

cc <- rxSparkConnect(reset = TRUE)
colInfo <- list(
    ArrDelay = list(
        type = "numeric"),
    DayOfWeek = list(
        type = "factor",
        levels = c(
            "Monday",
            "Sunday",
            "Wednesday",

kenthzhang

cc <- rxSparkConnect(reset = TRUE)
hdfsFileSystem <- RxHdfsFileSystem()
colInfo <- list(
    DayOfWeek = list(
        type = "factor",
        levels = c(
            "Monday",
            "Tuesday",
            "Wednesday",
            "Thursday",

kenthzhang

cc <- rxSparkConnect(reset = TRUE)
hdfsFileSystem <- RxHdfsFileSystem()
xdfData <- RxXdfData(file = "/share/AirlineDemoSmall/AirlineDemoSmallXDF", fileSystem = hdfsFileSystem)
.Message <- function(keys, data)
{
    df <- rxImport(data)
    rows <- nrow(df)
    if (rows > 80000)
    {
        warning("rows > 80000")

kenthzhang

cc <- rxSparkConnect(reset = TRUE)
hdfsFileSystem <- RxHdfsFileSystem()
textData <- RxTextData(file = "/share/AirlineDemoSmall/AirlineDemoSmall.csv", missingValueString = "M", stringsAsFactors = TRUE, fileSystem = hdfsFileSystem)
.Summary <- function(keys, data)
{
    df <- rxImport(data)
    nrow(df)
}

# single key

kenthzhang

# Create myDir in hdfs and copy csv file
source <-system.file("SampleData/AirlineDemoSmall.csv", package="RevoScaleR")
myDir <- "/share/AirlineDemoSmall"
rxHadoopMakeDir(myDir)
rxHadoopCopyFromLocal(source, myDir)

cc <- rxSparkConnect(reset = TRUE)

# Summary of data
hdfsFileSystem <- RxHdfsFileSystem()

kenthzhang