Read and Write Tables From Impala With R

library(DBI)

con <- dbConnect(odbc::odbc(),
                      Driver   = "Cloudera ODBC Driver for Impala 64-bit",
                      Host     = "dn1.pX.company.prod.saagie.io",
                      Port     = 21050,
                      Schema   = "default",
                      AuthMech = 3,
                      UseSASL  = 1,
                      UID      = "user",
                      PWD      = "password")

library(DBI)
library(getPass)

# Method 1 (interactive) : Use in RStudio. Enter your password in an interactive pop-up window.
system('kinit user',input=getPass('Enter your password: '))

# Method 2 (scripts) : Use outside of RStudio.
# Write the password in the command line or store it in an environment variable.
# Uncomment the following line to use it.
# system('echo password | kinit user')

con <- dbConnect(odbc::odbc(),
                 Driver = "Cloudera ODBC Driver for Impala 64-bit",
                 Host = "dn1.pX.company.prod.saagie.io",
                 Port = 21050,
                 Schema = "default",
                 AuthMech = 1)

# Troubleshooting: If you only see the first letter of each table, see "Fix Table Display Names in RStudio and dbListTables" below.
# Another workaround is to use "show schemas" and "show tables" in plain SQL.

# To list all tables of all schemas.
dbListTables(con)
dbGetQuery(con, 'show schemas')
dbGetQuery(con, 'show tables')

sample <- DBI::dbReadTable(con, "sample") # Fetch all data from a table

# To create a lazy "tbl" from an Impala table.
forecast_db <- tbl(con, in_schema('forecast', 'forecast_weekly'))
forecast_db # Print columns and column types

# The query is written in dplyr syntax, but is executed on a remote SQL database.
query <- forecast_db %>%
    summarise(mean_forecast = forecast %>% as.numeric() %>% mean)

show_query(query) # To show the query that will be executed.
query # To execute the lazy query when the result is printed.

# Example of verbs used by dplyr.
forecast_db %>%
    filter(prediction_date == max(prediction_date)) %>%
    group_by(reference) %>%
    summarise(forecast_mean = mean(forecast),
              forecast_max = max(forecast),
              nb_forecast_total = n())

sales_db <- tbl(con, in_schema('forecast', 'sales_weekly'))
sales_db

forecast_db <- forecast_db %>%
    mutate(reference = as.character(reference))

diff_db <- inner_join(forecast_db, sales_db, by=c('reference', 'year', 'week'))

diff_by_ref <- diff_db %>%
    group_by(reference, year, week) %>%
    summarise(diff_by_week = abs(forecast - quantity)) %>% # To differentiate between forecast and reality for each prediction.
    group_by(reference) %>%
    summarise(diff_by_ref = sum(diff_by_week)) # Sum of all differences for each reference.

diff_db # To execute all the lazy queries above.

#' To write data from R to Impala using HDFS. (1)
#'
#' @param con : The Impala connexion.
#' @param name : The name of the table created in Impala.
#' @param obj : The R DataFrame.
#' @param database : The Impala database to use.
#' @param overwrite : Overwrites the current table if it already exists.
#'
fastWriteTable <- function(con, name, obj, database = 'default', overwrite = F) {
  library(DBI)
  library(stringi)
  library(httr)
  library(data.table)

  if(!overwrite) {
    showRequest <- paste0("show tables in `", database, "` like '", name, "'")
    exists <- dbGetQuery(con, showRequest)
    if(nrow(exists) > 0) {
      print(paste0('Table ', database, '.', name, ' already exists. Use overwrite = T to overwrite table'))
      return('Table already exists')
    }
  }

  # HDFS information.
  hdfsUrl <- "https://nn1.pX.companyName.saagie.io:50470/webhdfs/v1"
  hdfsUser <- '&user.name=impala'

  # Informations about the CSV table.
  tableCsv <- paste0('tmp__', name, '__tmp') # Name of the table in Impala
  location <- paste0('/tmp/', tableCsv, '/') # Folder location of the csv file
  separator <- ';' # Separator in the csv file
  nameCsv <- paste0(name, '.csv')

  # To select the database.
  dbGetQuery(con, paste0('use ', database))

  # Column names and classes.
  cl <- sapply(obj, class)
  n <- names(cl)

  # To match R and Impala column types.
  r_types <- c('numeric', 'integer', 'character', 'factor', 'c(\"POSIXct\", \"POSIXt\")')
  impala_types <- c('DOUBLE', 'INTEGER', 'STRING', 'STRING', 'TIMESTAMP')

  # To replace R types by Impala types.
  cl <- stringi::stri_replace_all_fixed(cl, r_types, impala_types, vectorize_all=F)

  # To create each column as correct type.
  namesTypes <- paste0(dbQuoteIdentifier(con, make.names(n)), ' ', cl, collapse = ', ')

  dropRequest <- paste0("DROP TABLE IF EXISTS `", tableCsv, "`")
  dbGetQuery(con, dropRequest)

  # To write the CSV to disk and copy it to Impala.
  fwrite(obj, nameCsv, col.names = F, sep = ';')
  set_config( config( ssl_verifypeer = 0L ) )
  uri <- paste0(hdfsUrl, location, nameCsv, '?op=CREATE&overwrite=true', hdfsUser)
  response <- PUT(uri)
  stop_for_status(response)
  uriWrite <- response$url
  response <- PUT(uriWrite, body = upload_file(nameCsv))
  stop_for_status(response)

  # To create a table on the CSV file. (2)
  createRequest <- paste0("CREATE TABLE `", tableCsv, "` (", gsub('.', '_', namesTypes, fixed = T), ") row format delimited fields terminated by '", separator, "' stored as textfile location '", location, "'")
  dbGetQuery(con, createRequest)

  dropRequest2 <- paste0("DROP TABLE IF EXISTS `", name, "`")
  dbGetQuery(con, dropRequest2)

  # To copy the CSV table into a "parquet" table. (3)
  parquetRequest <- paste0("CREATE TABLE `", name, "` STORED AS PARQUET AS SELECT * FROM `", tableCsv, "`")
  dbGetQuery(con, parquetRequest)

  # To optimize the table.
  computeRequest <- paste0("COMPUTE STATS `", name, "`")
  dbGetQuery(con, computeRequest)

  # To remove the temporary CSV table. (4)
  dropTmpRequest <- paste0("DROP TABLE `", tableCsv, "`")
  dbGetQuery(con, dropTmpRequest)

  return('Table correctly uploaded')
}

# To list all databases.
dbListTables(conn)
# To list the "default" database.
dbGetQuery(conn,"show tables")

# To get all the elements of the "default" database.
d <- dbReadTable(conn,"table_name")
# OR
d <- dbGetQuery(conn,"select * from table_name")
# To get all the other elements that are not in the "default" database.
d <- dbReadTable(conn, nameBDD.table_name)

dbSendUpdate(conn, "CREATE TABLE table_name (attribute1 string, attribute2 int) STORED AS PARQUET")

dbGetQuery(conn, "INSERT INTO table_name VALUES ('test', 1)")

dbGetQuery(conn, "INVALIDATE METADATA")