hrbrmstr / sergeant Goto Github PK

library(sergeant)

dc <- drill_connection() 

dc %>% drill_query('SELECT full_name, salary FROM cp.`employee.json` LIMIT 3')
#> # A tibble: 3 x 2
#>         full_name salary
#> *           <chr>  <dbl>
#> 1    Sheri Nowmer  80000
#> 2 Derrick Whelply  40000
#> 3  Michael Spence  40000

dc %>% drill_query('SELECT full_name, salary FROM cp.`employee.json` LIMIT 3;')
#> Query ==> SELECT full_name, salary FROM cp.`employee.json` LIMIT 3;
#> PARSE ERROR: Encountered ";" at line 1, column 57.
#> Was expecting one of:
#>     <EOF> 
#>     "OFFSET" ...
#>     "FETCH" ...
#>     
#> 
#> SQL Query SELECT full_name, salary FROM cp.`employee.json` LIMIT 3;
#>                                                         ^
#> 
#> 
#> [Error Id: a172e1a9-833c-4d89-9b83-de4e250cd1c5 on 192.168.1.3:31010]

#install.packages("sergeant")
library(sergeant)
#> Loading required package: DBI
#> Loading required package: dplyr
#> 
#> Attaching package: 'dplyr'
#> The following objects are masked from 'package:stats':
#> 
#>     filter, lag
#> The following objects are masked from 'package:base':
#> 
#>     intersect, setdiff, setequal, union
#> Loading required package: dbplyr
#> 
#> Attaching package: 'dbplyr'
#> The following objects are masked from 'package:dplyr':
#> 
#>     ident, sql
library(tidyverse)

# drill server IP Address
drill_ip <- "xxx.xxx.xxx.xxx"
drill_jdbc_port <- "31010" 

# Connecting to drill server
# src_drill: dplyr connection
# drill_connection: REST API connection
# drill_jdbc: JDBC connection (if remote, needs zk etc)

# Connect using dplyr
ds_dp <- src_drill(drill_ip) 
tbl(ds_dp, "cp.`employee.json`")
#> # Source:   table<cp.`employee.json`> [?? x 16]
#> # Database: DrillConnection
#>    store_id gender department_id birth_date supervisor_id last_name
#>       <dbl> <chr>          <dbl> <date>             <dbl> <chr>    
#>  1        0 F                  1 1961-08-26             0 Nowmer   
#>  2        0 M                  1 1915-07-03             1 Whelply  
#>  3        0 M                  1 1969-06-20             1 Spence   
#>  4        0 F                  1 1951-05-10             1 Gutierrez
#>  5        0 F                  2 1942-10-08             1 Damstra  
#>  6        0 F                  3 1949-03-27             1 Kanagaki 
#>  7        9 F                 11 1922-08-10             5 Brunner  
#>  8       21 F                 11 1979-06-23             5 Blumberg 
#>  9        0 M                  5 1949-08-26             1 Stanz    
#> 10        1 M                 11 1967-06-20             5 Murraiin 
#> # ... with more rows, and 10 more variables: position_title <chr>,
#> #   hire_date <dttm>, management_role <chr>, salary <dbl>,
#> #   marital_status <chr>, full_name <chr>, employee_id <dbl>,
#> #   education_level <chr>, first_name <chr>, position_id <dbl>
tbl(ds_dp, "SELECT * FROM dfs.`c:/apps/apache-drill-1.14.0/sample-data/nation.parquet` LIMIT 5")
#> Error in readr::type_convert(out$rows): is.data.frame(df) is not TRUE

ds_rest <- drill_connection(drill_ip) 
drill_query(ds_rest,"SELECT * FROM dfs.`c:/apps/apache-drill-1.14.0/sample-data/nation.parquet` LIMIT 5")
#> Parsed with column specification:
#> cols(
#>   N_COMMENT = col_character(),
#>   N_NAME = col_character(),
#>   N_NATIONKEY = col_double(),
#>   N_REGIONKEY = col_double()
#> )
#> # A tibble: 5 x 4
#>   N_COMMENT            N_NAME    N_NATIONKEY N_REGIONKEY
#> * <chr>                <chr>           <dbl>       <dbl>
#> 1 haggle. carefully f  ALGERIA             0           0
#> 2 al foxes promise sly ARGENTINA           1           1
#> 3 y alongside of the p BRAZIL              2           1
#> 4 eas hang ironic, sil CANADA              3           1
#> 5 y above the carefull EGYPT               4           4

devtools::install_git("https://git.sr.ht/~hrbrmstr/sergeant", ref="0.8.0")

> db <- src_drill("localhost")
> fr24pos <- tbl(db, "(SELECT snapshot_id, longitude, latitude, altitude, heading, speed, CAST(REGEXP_REPLACE(`squawk`, '\r', '') AS int) AS `squawk`, radar_id FROM dfs.fr24.`/*.csv`)")
> fr24pos
# Source:   table<(SELECT snapshot_id, longitude, latitude, altitude, heading, speed,
', '') AS int) AS `squawk`, radar_id FROM
#   dfs.fr24.`/*.csv`)> [?? x 8]
# Database: DrillConnection
   altitude squawk snapshot_id heading latitude radar_id speed longitude
      <int>  <int>       <int>   <int>    <dbl>    <int> <int>     <dbl>
 1        0      0  1486340102      40     50.1     8520     0      14.3
 2        0      0  1486340568      58     50.1     8520     0      14.3
 3        0      0  1486340668     247     50.1     8520     0      14.3
 4        0      0  1486340773     232     50.1     8520     0      14.3
 5        0      0  1486340828      77     50.1     8520     0      14.3
 6        0      0  1486340903      41     50.1     8520     0      14.3
 7        0      0  1486341363      48     50.1     8520     0      14.3
 8        0      0  1486341828     241     50.1     8520     0      14.3
 9        0      0  1486342643     120     50.1     8520     0      14.3
10        0      0  1486342870     255     50.1     8520     0      14.3
# ... with more rows
> fr24pos %>% filter(altitude > 1500) %>% head()
# Source:   lazy query [?? x 8]
# Database: DrillConnection
  altitude squawk snapshot_id heading latitude radar_id speed longitude
     <int>  <int>       <int>   <int>    <dbl>    <int> <int>     <dbl>
1    39000  30276  1486339252     114     37.9      230   506      44.0
2    39000  30276  1486339315     114     37.8      230   509      44.2
3    38975  20836  1486339327     114     37.8      230   509      44.2
4    39025  20836  1486339361     114     37.8      230   508      44.2
5    39000  20836  1486339452     114     37.7      230   505      44.6
6    39000  20836  1486339513     131     37.6      230   503      44.7
> fr24pos %>% select(snapshot_id, longitude, latitude, heading, speed, everything()) %>% filter(altitude > 1500) %>% head()
# Source:   lazy query [?? x 8]
# Database: DrillConnection
  altitude squawk snapshot_id heading latitude radar_id speed longitude
     <int>  <int>       <int>   <int>    <dbl>    <int> <int>     <dbl>
1    39000  30276  1486339252     114     37.9      230   506      44.0
2    39000  30276  1486339315     114     37.8      230   509      44.2
3    38975  20836  1486339327     114     37.8      230   509      44.2
4    39025  20836  1486339361     114     37.8      230   508      44.2
5    39000  20836  1486339452     114     37.7      230   505      44.6
6    39000  20836  1486339513     131     37.6      230   503      44.7 

library(sergeant)
ncdr <- src_drill(host = "localhost", port=22023L)

# The connection appears to work
ncdr
#> src:  DrillConnection
#> tbls: INFORMATION_SCHEMA, cp.default, dfs.default, dfs.root, dfs.tmp, sys

#But I can't do a query
snaps <- drill_query(ncdr, "SELECT * FROM INFORMATION_SCHEMA")
#> Error: length(url) == 1 is not TRUE

# It looks like there's no URL to send the query to
sergeant:::make_server(ncdr)
#> character(0)

#This seems to be because `make_server` doesn't get the host url correctly
sergeant:::make_server
#> function (drill_con) 
#> {
#>     sprintf("%s://%s:%s", ifelse(drill_con$ssl[1], "https", "http"), 
#>         drill_con$host, drill_con$port)
#> }
#> <environment: namespace:sergeant>
ncdr$host
#> NULL
ncdr$con@host
#> [1] "localhost"

library(sergeant)
con <- src_drill()

fields_to_select <- c("first_name", "last_name")

glue::glue_sql("SELECT {`fields_to_select`*} FROM cp.`employee.json` LIMIT 20", .con = con)
#> Error in (function (classes, fdef, mtable) : unable to find an inherited method for function 'dbQuoteIdentifier' for signature '"src_drill", "character"'

library(sergeant)
library(dplyr)
#> 
#> Attaching package: 'dplyr'
#> The following objects are masked from 'package:stats':
#> 
#>     filter, lag
#> The following objects are masked from 'package:base':
#> 
#>     intersect, setdiff, setequal, union

ds <- src_drill()
emp_full <- tbl(ds, "cp.`employee.json`")
emp_full %>% select(full_name)
#> # Source:   lazy query [?? x 1]
#> # Database: DrillConnection
#>    full_name        
#>    <chr>            
#>  1 Sheri Nowmer     
#>  2 Derrick Whelply  
#>  3 Michael Spence   
#>  4 Maya Gutierrez   
#>  5 Roberta Damstra  
#>  6 Rebecca Kanagaki 
#>  7 Kim Brunner      
#>  8 Brenda Blumberg  
#>  9 Darren Stanz     
#> 10 Jonathan Murraiin
#> # ... with more rows

emp_partial <- tbl(ds, "SELECT full_name from cp.`employee.json`") #edits? we don't need no stinkin' edits!
emp_partial %>% select(full_name)
#> Error in .f(.x[[i]], ...): object 'full_name' not found

# Packages
library(sergeant)
#> Loading required package: DBI
#> Loading required package: dplyr
#> 
#> Attaching package: 'dplyr'
#> The following objects are masked from 'package:stats':
#> 
#>     filter, lag
#> The following objects are masked from 'package:base':
#> 
#>     intersect, setdiff, setequal, union
#> Loading required package: dbplyr
#> 
#> Attaching package: 'dbplyr'
#> The following objects are masked from 'package:dplyr':
#> 
#>     ident, sql

# Connect to Drill.
con <- drill_connection()

# Alternatively with drill_use.
drill_use(con, "dfs.tmp")
#> Parsed with column specification:
#> cols(
#>   summary = col_character(),
#>   ok = col_character()
#> )
#> Default schema changed to [dfs.tmp]true
drill_query(con, "CREATE TABLE employee as SELECT * FROM cp.`employee.json`")
#> Query ==> CREATE TABLE employee as SELECT * FROM cp.`employee.json`
#> VALIDATION ERROR: Root schema is immutable. Creating or dropping tables/views is not allowed in root schema.Select a schema using 'USE schema' command.
#> 
#> 
#> [Error Id: d4d341b3-07da-4562-813e-67f0ef89c028 on DESKTOP-73M5KKL.mshome.net:31010]

devtools::session_info()
#> Session info -------------------------------------------------------------
#>  setting  value                       
#>  version  R version 3.4.3 (2017-11-30)
#>  system   x86_64, mingw32             
#>  ui       RTerm                       
#>  language (EN)                        
#>  collate  English_United States.1252  
#>  tz       America/New_York            
#>  date     2018-03-01
#> Packages -----------------------------------------------------------------
#>  package    * version date       source        
#>  assertthat   0.2.0   2017-04-11 CRAN (R 3.4.3)
#>  backports    1.1.2   2017-12-13 CRAN (R 3.4.3)
#>  base       * 3.4.3   2017-12-06 local         
#>  bindr        0.1     2016-11-13 CRAN (R 3.4.3)
#>  bindrcpp     0.2     2017-06-17 CRAN (R 3.4.3)
#>  colorspace   1.3-2   2016-12-14 CRAN (R 3.4.3)
#>  compiler     3.4.3   2017-12-06 local         
#>  curl         3.1     2017-12-12 CRAN (R 3.4.3)
#>  datasets   * 3.4.3   2017-12-06 local         
#>  DBI        * 0.7     2017-06-18 CRAN (R 3.4.3)
#>  dbplyr     * 1.2.1   2018-02-19 CRAN (R 3.4.3)
#>  devtools     1.13.5  2018-02-18 CRAN (R 3.4.3)
#>  digest       0.6.15  2018-01-28 CRAN (R 3.4.3)
#>  dplyr      * 0.7.4   2017-09-28 CRAN (R 3.4.3)
#>  evaluate     0.10.1  2017-06-24 CRAN (R 3.4.3)
#>  glue         1.2.0   2017-10-29 CRAN (R 3.4.3)
#>  graphics   * 3.4.3   2017-12-06 local         
#>  grDevices  * 3.4.3   2017-12-06 local         
#>  hms          0.4.1   2018-01-24 CRAN (R 3.4.3)
#>  htmltools    0.3.6   2017-04-28 CRAN (R 3.4.3)
#>  httr         1.3.1   2017-08-20 CRAN (R 3.4.3)
#>  jsonlite     1.5     2017-06-01 CRAN (R 3.4.3)
#>  knitr        1.20    2018-02-20 CRAN (R 3.4.3)
#>  magrittr     1.5     2014-11-22 CRAN (R 3.4.3)
#>  memoise      1.1.0   2017-04-21 CRAN (R 3.4.3)
#>  methods    * 3.4.3   2017-12-06 local         
#>  munsell      0.4.3   2016-02-13 CRAN (R 3.4.3)
#>  pillar       1.1.0   2018-01-14 CRAN (R 3.4.3)
#>  pkgconfig    2.0.1   2017-03-21 CRAN (R 3.4.3)
#>  plyr         1.8.4   2016-06-08 CRAN (R 3.4.3)
#>  purrr        0.2.4   2017-10-18 CRAN (R 3.4.3)
#>  R6           2.2.2   2017-06-17 CRAN (R 3.4.3)
#>  Rcpp         0.12.15 2018-01-20 CRAN (R 3.4.3)
#>  readr        1.1.1   2017-05-16 CRAN (R 3.4.3)
#>  rlang        0.2.0   2018-02-20 CRAN (R 3.4.3)
#>  rmarkdown    1.8     2017-11-17 CRAN (R 3.4.3)
#>  rprojroot    1.3-2   2018-01-03 CRAN (R 3.4.3)
#>  scales       0.5.0   2017-08-24 CRAN (R 3.4.3)
#>  sergeant   * 0.5.2   2017-07-17 CRAN (R 3.4.3)
#>  stats      * 3.4.3   2017-12-06 local         
#>  stringi      1.1.6   2017-11-17 CRAN (R 3.4.2)
#>  stringr      1.3.0   2018-02-19 CRAN (R 3.4.3)
#>  tibble       1.4.2   2018-01-22 CRAN (R 3.4.3)
#>  tools        3.4.3   2017-12-06 local         
#>  utils      * 3.4.3   2017-12-06 local         
#>  withr        2.1.1   2017-12-19 CRAN (R 3.4.3)
#>  yaml         2.1.16  2017-12-12 CRAN (R 3.4.3)

> library(sergeant)
> tbl(src_drill('localhost'), 'cp.`employee.json`')
Request failed [500]. Retrying in 1 seconds...
Request failed [500]. Retrying in 2 seconds...
Request failed [500]. Retrying in 1.8 seconds...
Request failed [500]. Retrying in 1.7 seconds...
# Source:   table<cp.`employee.json`> [?? x 8]
# Database: DrillConnection
Warning message:
PARSE ERROR: Encountered "FROM \"" at line 1, column 1.
SQL Query: SELECT *
FROM "cp.`employee.json`"
^
LIMIT 11

Original Query:

  1: SELECT *
  2: FROM "cp.`employee.json`"
  3: LIMIT 11

Query Profile Error Link:
http://localhost:8047/profiles/772b3c22-8d04-43d5-b733-9e2ae0bcbb50
> con <- drill_connection('localhost')

> drill_query(
+     con,
+     'SELECT *
+      FROM "cp.`employee.json`"
+      LIMIT 11'
+ )
  |======================================================================| 100%
Request failed [500]. Retrying in 1.4 seconds...
  |======================================================================| 100%
Request failed [500]. Retrying in 1.9 seconds...
  |======================================================================| 100%
Query ==> SELECT *      FROM "cp.`employee.json`"      LIMIT 11
PARSE ERROR: Encountered "FROM \"" at line 1, column 6.

SQL Query: SELECT *
     FROM "cp.`employee.json`"
     ^
     LIMIT 11

[Error Id: dc495019-3219-4b88-b5f8-9a4fd657dc0f ]

> drill_query(
+     con,
+     'SELECT *
+      FROM cp.`employee.json`
+      LIMIT 11'
+ )
  |======================================================================| 100%
# A tibble: 11 x 16
   employee_id full_name first_name last_name position_id position_title
         <dbl> <chr>     <chr>      <chr>           <dbl> <chr>
 1           1 Sheri No… Sheri      Nowmer              1 President
 2           2 Derrick … Derrick    Whelply             2 VP Country Ma…
 3           4 Michael … Michael    Spence              2 VP Country Ma…
 4           5 Maya Gut… Maya       Gutierrez           2 VP Country Ma…
 5           6 Roberta … Roberta    Damstra             3 VP Informatio…
 6           7 Rebecca … Rebecca    Kanagaki            4 VP Human Reso…
 7           8 Kim Brun… Kim        Brunner            11 Store Manager
 8           9 Brenda B… Brenda     Blumberg           11 Store Manager
 9          10 Darren S… Darren     Stanz               5 VP Finance
10          11 Jonathan… Jonathan   Murraiin           11 Store Manager
11          12 Jewel Cr… Jewel      Creek              11 Store Manager
# … with 10 more variables: store_id <dbl>, department_id <dbl>,
#   birth_date <chr>, hire_date <chr>, salary <dbl>, supervisor_id <dbl>,
#   education_level <chr>, marital_status <chr>, gender <chr>,
#   management_role <chr>

> devtools::install_github("hrbrmstr/sergeant")
Downloading GitHub repo hrbrmstr/sergeant@master
Error in structure(.External(.C_dotTclObjv, objv), class = "tclObj") : 
  [tcl] bad pad value "2m": must be positive screen distance.

> installed.packages() %>% as_tibble() %>% filter(Package == "sergeant") %>% flatten() %>% .$Version
[1] "0.5.2"
> library(sergeant)
> db <- src_drill("localhost")
> drill_connection() %>% drill_version()
[1] "1.16.0"

> sessionInfo()
R version 3.6.0 (2019-04-26)
Platform: x86_64-pc-linux-gnu (64-bit)
Running under: Ubuntu 18.04.2 LTS

Matrix products: default
BLAS:   /usr/lib/x86_64-linux-gnu/openblas/libblas.so.3
LAPACK: /usr/lib/x86_64-linux-gnu/libopenblasp-r0.2.20.so

locale:
 [1] LC_CTYPE=pt_BR.UTF-8       LC_NUMERIC=C              
 [3] LC_TIME=pt_BR.UTF-8        LC_COLLATE=en_US.UTF-8    
 [5] LC_MONETARY=pt_BR.UTF-8    LC_MESSAGES=en_US.UTF-8   
 [7] LC_PAPER=pt_BR.UTF-8       LC_NAME=C                 
 [9] LC_ADDRESS=C               LC_TELEPHONE=C            
[11] LC_MEASUREMENT=pt_BR.UTF-8 LC_IDENTIFICATION=C       

attached base packages:
[1] stats     graphics  grDevices utils     datasets  methods   base     

other attached packages:
[1] nvimcom_0.9-82 colorout_1.2-1 usethis_1.5.0  devtools_2.0.2

loaded via a namespace (and not attached):
 [1] Rcpp_1.0.1        magrittr_1.5      pkgload_1.0.2     R6_2.4.0         
 [5] rlang_0.3.4       tcltk_3.6.0       tools_3.6.0       pkgbuild_1.0.3   
 [9] parallel_3.6.0    sessioninfo_1.1.1 cli_1.1.0         withr_2.1.2      
[13] remotes_2.0.4     assertthat_0.2.1  rprojroot_1.3-2   digest_0.6.18    
[17] crayon_1.3.4      processx_3.3.1    callr_3.2.0       fs_1.3.1         
[21] ps_1.3.0          curl_3.3          testthat_2.1.1    memoise_1.1.0    
[25] glue_1.3.1        compiler_3.6.0    desc_1.2.0        backports_1.1.4  
[29] prettyunits_1.0.2

library(sparklyr)
library(dplyr)

# first install spark 2.40 hadoop 2.7 with sparklyr::spark_install()

Sys.setenv("SPARK_MEM" = "12g")

config <- spark_config()
config$`sparklyr.shell.driver-memory` <- '12G'
config$`sparklyr.shell.executor-memory` <- '4G'
config$sparklyr.defaultPackages <- "com.datastax.spark:spark-cassandra-connector_2.11:2.0.0-M3"
config$spark.cassandra.cassandra.host <- "localhost"
config$spark.driver.maxResultSize <- "4G"
config$spark.executor.cores <- 3

# is pushdown option TRUE?

sc <- spark_connect(master = "local", config = config)

# for this connection, load all records

system.time(
  spark_read_csv(sc, memory = FALSE,
    name = "artdata", path = "file:///home/roger/artdata/artdata.tsv", delimiter = "\t")
)

#user   system  elapsed 
#6.154    7.060 1559.874 

# generate a parquet file based on the dataframe above

system.time(
  spark_write_parquet(
    tbl(sc, "artdata"),  
    "file:///home/roger/artdata/artdata.parquet")
)

#user   system  elapsed 
#14.634   16.586 3816.375 

# the parquet-file is 3.8 GB on disk, smaller than the zip

spark_tbl_handle <- spark_read_parquet(sc, memory = FALSE,
  "artdata", "file:///home/roger/artdata/artdata.parquet")


has_valid_bor <- function() {
  
  bor <- 
    spark_tbl_handle %>%
    count(basisOfRecord) %>%
    collect() %>%
    mutate(is_ok = basisOfRecord %in% c(
        "humanobservation", 
        "machineobservation"
      )
    )

  bor %>% pull(is_ok) %>% all
}

n_rowcount <- function() {
  
  spark_tbl_handle %>%
  summarise(n = n()) %>%
  pull(n)

}


has_valid_id <- function() {
  
  ids <- 
    spark_tbl_handle %>%
    count(occurrenceID) %>%
    filter(n > 1, is.na(occurrenceID)) %>%
    collect()
  
  nrow(ids) == 0
  
}


system.time(
  has_valid_bor()
)

system.time(
  has_valid_id()
)

system.time(
  n_rowcount()
)

sort_artdata <- function() {
  spark_tbl_handle %>%
  arrange(occurrenceID) %>%
  head(10) %>%
  collect()
}

system.time(
  sort_artdata()
)

# sorting in spark takes about 5 minutes...
#user  system elapsed 
#3.182   1.370 282.698

library(sergeant)
library(tidyverse)

if (Sys.which("docker") == "")
  stop("Please install docker first - see https://docs.docker.com/install/")

# install and run official Apache Drill software
system("docker stop drill; docker rm drill; docker run -i --name drill -e DRILL_HEAP=10G -v /home/markus/tmp/artdata:/tmp -p 8047:8047 --detach drill/apache-drill:1.16.0 /bin/bash")

dc <- drill_connection("localhost") 

df <- drill_query(dc, "SELECT occurrenceId FROM dfs.`/tmp/artdata.parquet`")