Interoperable analysis pipelines

In analysisPipelines: Compose Interoperable Analysis Pipelines & Put Them in Production

Introduction

This vignette explains how interoperable pipelines containing functions operating on different engines such as R, Spark and Python can be configured and executed through the analysisPipelines package. Currently, the package supports interoperable pipelines containing R and Spark batch functions.

If the package is new to you, it is recommended that you go through the Analysis pipelines - Core functionality and working with R data frames and functions vignette first.

Important Note

Using Spark as an engine requires the SparkR package to be installed. SparkR is distributed natively with Apache Spark and is not distributed on CRAN.

library(analysisPipelines)
knitr::opts_chunk$set(
    eval = FALSE
  )

An example of an interoperable pipeline

In this vignette we demonstrate an interoperable pipeline built using the analysisPipelines package, which contains a couple of filtering/ aggregation functions performed in Spark, which is then subsequently visualized through R functions using ggplot2

Initializing a Spark connection from R and loading the data

We initialize a Spark session using the sparkRSessionCreateIfNotPresent helper function in the analysisPipelines package, which internally uses SparkR. We then read the data into the Spark session using functions in the SparkR package. In this case we read a .csv file, though SparkR can work with multiple other data sources

## Define these variables as per the configuration of your machine. This is just an example.
sparkHome <- "/Users/naren/softwares/spark-2.3.1-bin-hadoop2.7/"
sparkMaster <- "local[1]"
sparkPackages <- c("org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.1")

sparkRSessionCreateIfNotPresent(sparkHome = sparkHome, master = sparkMaster, sparkPackages = sparkPackages)

inputDataset <- iris

# Replacing '.' in column names with '_' as SparkR is not able to deal with '.' in column names
colnames(inputDataset) <- gsub(".", "_", colnames(inputDataset), fixed = T)

Initializing Python connection

## Define these variables as per the configuration of your machine. This is just an example.

analysisPipelines::setPythonEnvir('python', '/Users/naren/anaconda3/bin/python')
os <- reticulate::import("os")
numpy <- reticulate::import("numpy")
pandas <- reticulate::import("pandas")
sklearn <- reticulate::import("sklearn")

reticulate::source_python(system.file("python/sampleFunctions.py", package = "analysisPipelines"))

reticulate::py_config()

Creating an analysisPipeline object

We then initialize an AnalysisPipeline, with the input dataset

pipelineObj <- AnalysisPipeline(input = inputDataset)

Registering functions to work in the Spark environment

In order to manipulate the data in the Spark environment, we define our own functions using SparkR interface functions. We then register these functions with the AnalysisPipeline object, so that they can be used in constructing a pipeline.

The getRegistry function lists all the registered functions, along with details such as which engine they should run on.

getSchema <- function(inputDataset) {
  sparkSchema <- SparkR::schema(inputDataset)
 return(sparkSchema)
}

filterData <- function(inputDataset, condition) {
  filteredData <- SparkR::filter(inputDataset, condition)
 return(filteredData)
}

registerFunction(functionName = "getSchema", engine = "spark") 
registerFunction(functionName = "filterData", engine = "spark") 


getRegistry()

Registering R functions

Similar to the Spark functions, we register some user-defined functions in R. In this case to plot a bivariate plot using ggplot2.

rBivarPlots <- function(dataset, select_var_name_1, select_var_name_2, priColor = "blue", secColor= "black") {

  numeric_cols <- unlist(getDatatype(dataset)['numeric_cols'])
  cat_cols <- unlist(getDatatype(dataset)['cat_cols'])

  if (select_var_name_1 %in% numeric_cols && select_var_name_2 %in% numeric_cols) {
    x = dataset[, select_var_name_1]
    y = dataset[, select_var_name_2]
    bivarPlot <-
      ggplot2::ggplot(dataset, ggplot2::aes(x, y)) +
      ggplot2::geom_point(color = priColor, alpha = 0.7) +
      ggplot2::geom_smooth(method = lm, color = secColor) +
      ggplot2::xlab(select_var_name_1) +
      ggplot2::ylab(select_var_name_2) + ggplot2::theme_bw() +
      ggplot2::ggtitle(paste(
        'Bivariate plot for',
        select_var_name_1,
        'and',
        select_var_name_2,
        sep = ' '
      )) +
      ggplot2::theme(
        plot.title = ggplot2::element_text(hjust = 0.5, size = 10),
        axis.text = ggplot2::element_text(size = 10),
        axis.title = ggplot2::element_text(size = 10)
      )



  } else if (select_var_name_1 %in% cat_cols &&
             select_var_name_2 %in% cat_cols) {
    new_df <- dataset %>% dplyr::group_by_(.dots=c(select_var_name_1,select_var_name_2)) %>% dplyr::summarise(n = dplyr::n())
    colfunc <- grDevices::colorRampPalette(c(priColor, "white" , secColor))
    colorvar <- length(unique(new_df[[select_var_name_2]]))
    a=as.vector(as.character(unique(new_df[[select_var_name_1]])))
    y=new_df[[select_var_name_1]]
    label=new_df[[select_var_name_2]]
    bivarPlot <-ggplot2::ggplot(new_df, ggplot2::aes(x = y, y= n, fill = label)) +
      ggplot2::geom_bar(position = "dodge", stat = "identity",alpha=0.9) +
      ggplot2::guides(fill=ggplot2::guide_legend(title=select_var_name_2)) +
      ggplot2::coord_flip()+
      ggplot2::xlab(select_var_name_1) +
      ggplot2::ylab("count") + ggplot2::theme_bw() +
      ggplot2::ggtitle(paste('Bivariate plot for',select_var_name_1,'and',select_var_name_2,sep=' '))+
      ggplot2::theme(plot.title = ggplot2::element_text(hjust = 0.5, size = 10),axis.text = ggplot2::element_text(size=10),
                     axis.title=ggplot2::element_text(size=10),legend.position="bottom",axis.text.x=ggplot2::element_text(angle=45, hjust=1))+ ggplot2::scale_fill_manual(values = colfunc(colorvar))


  } else {
    cols <- c(select_var_name_1, select_var_name_2)
    cat_col <- cols[which(cols %in% cat_cols)]
    num_col <- cols[which(cols %in% numeric_cols)]
    a = as.vector(as.character(unique(dataset[[cat_col]])))
    y = dataset[[cat_col]]
    x = dataset[[num_col]]
    bivarPlot <-
      ggplot2::ggplot(dataset, ggplot2::aes(x = y, y = x)) +
      ggplot2::geom_point(color = priColor, alpha = 0.7) +
      ggplot2::coord_flip() +
      ggplot2::xlab(cat_col) +
      ggplot2::ylab(num_col) + ggplot2::theme_bw() +
      ggplot2::ggtitle(paste(
        'Bivariate plot for',
        select_var_name_1,
        'and',
        select_var_name_2,
        sep = ' '
      )) +
      ggplot2::theme(
        plot.title = ggplot2::element_text(hjust = 0.5, size = 10),
        axis.text = ggplot2::element_text(size = 10),
        axis.title = ggplot2::element_text(size = 10)
      )
  }

  return(bivarPlot)
}

registerFunction(functionName = "rBivarPlots", engine = "r", heading = "Bivariate analysis")

getRegistry()

Registering Python functions

registerFunction("decisionTreeTrainAndTest", engine = "python", isDataFunction = F, firstArgClass = "numpy.ndarray")
getRegistry()

Interoperable pipeline containing R, Spark and Python functions

• Here we consider a typical use case of performing data filtering/ aggregations and so on and Spark, and then using R to visualize, and Python to run a Machine learning model

We first visualize the data without filtering:

pipelineObj %>>% rBivarPlots(select_var_name_1 = "Sepal_Length", select_var_name_2 =  "Sepal_Width", 
                     priColor = "blue", secColor = "green", storeOutput = T) -> vizWithoutFilterPipeline
vizWithoutFilterPipeline %>>% getPipeline
vizWithoutFilterPipeline %>>% assessEngineSetUp
vizWithoutFilterPipeline %>>% generateOutput -> opWithoutFilter
opWithoutFilter %>>% getOutputById(1)

We then perform filtering on one of the variables in Spark, before visualizing in R

pipelineObj %>>% filterData_spark(condition = "Species == 'setosa'") %>>% 
  rBivarPlots(select_var_name_1 = "Sepal_Length", select_var_name_2 =  "Sepal_Width", 
                     priColor = "blue", secColor = "green", outAsIn = T, storeOutput = T) -> singleFilterPipeline
singleFilterPipeline %>>% visualizePipeline

singleFilterPipeline %>>% generateOutput -> opWithFilter
opWithFilter %>>% getOutputById(2)

Finally, we show a case, where sequential filtering steps are performed in Spark, before visualizing in R, and running a decision tree model in Python.

Note, that in this case, getTargetForPyClassifcation and getTargetForPyClassification have been registered as data functions. Type conversions between R, Spark and Python for data functions are performed automatically by the package.

pipelineObj %>>% filterData_spark(condition = "Species == 'setosa' or Species == 'virginica'") %>>% 
  filterData_spark(condition = "Petal_Length > 3.7", outAsIn = T) %>>%
  rBivarPlots(select_var_name_1 = "Sepal_Length", select_var_name_2 =  "Sepal_Width", 
                     priColor = "blue", secColor = "green", outAsIn = T, storeOutput = T) %>>%
  getFeaturesForPyClassification(dataset = ~f2, featureNames = c("Sepal_Length",
                                                                 "Sepal_Width",
                                                                 "Petal_Length")) %>>% 
          getTargetForPyClassification(dataset = ~f2, targetVarName = "Species", positiveClass = "setosa") %>>%
          decisionTreeTrainAndTest_py(data = ~f4, target = ~f5, newData = ~f4, storeOutput = T) -> twoFilterPipeline

twoFilterPipeline %>>% visualizePipeline

twoFilterPipeline %>>% generateOutput -> opWith2Filters
opWith2Filters %>>% getOutputById(3)
opWith2Filters %>>% getOutputById(6)

Supplementary Note

The analysisPipelines package internally uses the SparkR package to interface with Spark. SparkR masks many typical data manipulation and processing functions from base as well as packages like dplyr. Therefore, ensure you use function scoping when calling a function.

analysisPipelines documentation built on July 1, 2020, 7:09 p.m.