Hive CLI および/または beeline CLI および/または Spark (2.3.1) WITH から実行できる select および join ステートメントを作成しましたenableHiveSupport=TRUE。(注: API に SparkR を使用しています)
beeline を使用した結合と書き込みには 30 分かかりますが、Spark with を使用した結合と書き込みにenableHiveSupport=TRUEは 3.5 時間かかります。これは、Spark とそのコネクタがくだらないことを意味するか、または私が本来あるべき方法で Spark を使用していないことを意味します...そして、Spark の「スライスされたパン以来の最高のもの」の解説について読んだすべては、おそらくそれを正しく使用していないことを意味します.
Hive テーブルから読み取りたいのですが、Hive に何もしてほしくありません。毎月のデータに対して結合を実行し、各レコードの毎月のデルタで回帰を実行してから、必要に応じて、Hive から読み取り可能な寄木細工の出力テーブルに最終的な勾配/ベータを出力したいと思います...できれば同じ方法でパーティション分割しますHive からの入力データとして使用しているテーブルを分割しました。
リクエストに応じて、コードをいくつか示します...しかし、何も学べないと思います。ビッグ データ クエリでは、再現可能な結果は得られません。
Sys.setenv(SPARK_HOME="/usr/hdp/current/spark2-client")
sessionInfo()
library(SparkR, lib.loc = c(file.path(Sys.getenv("SPARK_HOME"), "R", "lib")))
sparkR.stop()
Sys.setenv(SPARKR_SUBMIT_ARGS="--master yarn sparkr-shell") #--master yarn-client sparkr-shell
Sys.setenv(LOCAL_DIRS="/tmp")
config = list()
config$spark.cores.max <- 144L
config$spark.executor.cores <- 2L
config$spark.executor.memory <- '8g'
config$spark.driver.cores <- 6L
config$spark.driver.maxResultSize <-"0"
config$spark.driver.memory <- "32g"
config$spark.shuffle.service.enabled<-TRUE
config$spark.dynamicAllocation.enabled <-FALSE
config$spark.scheduler.mode <- 'FIFO'
config$spark.ui.port<-4044L
sparkR.session(master = "yarn",
sparkHome = Sys.getenv("SPARK_HOME"),
sparkConfig = config,
enableHiveSupport = TRUE)
print("Connected!")
############ SET HIVE CONFIG
collect(sql("SET hive.exec.dynamic.partition") )
sql("SET hive.exec.dynamic.partition=true")
collect(sql("SET hive.exec.dynamic.partition.mode"))
sql("SET hive.exec.dynamic.partition.mode=nonstrict")
##
start_time <- Sys.time()
############### READ IN DATA {FROM HIVE}
sql('use historicdata')
data_tables<-collect(sql('show tables'))
exporttabs <- grep(pattern = 'export_historic_archive_records',x = data_tables$tableName,value = TRUE)
jointabs<-sort(exporttabs)[length(exporttabs)-(nMonths-1):0]
currenttab<-jointabs[6]
############### CREATE TABLE AND INSERT SCRIPTS
sql(paste0('use ',hivedb))
sql(paste0('DROP TABLE IF EXISTS histdata_regression',tab_suffix))
sSelect<-paste0("Insert Into TABLE histdata_regression",tab_suffix," partition (scf) SELECT a.idkey01, a.ssn7")
sCreateQuery<-paste0("CREATE TABLE histdata_regression",tab_suffix," (idkey01 string, ssn7 string")
sFrom<-paste0("FROM historicdata.",jointabs[nMonths]," a")
sAlias<-letters[nMonths:1]
DT <- gsub(pattern = "export_historic_archive_records_",replacement = "",jointabs)
DT<-paste0(DT)
for (i in nMonths:1) {
sSelect<-paste0(sSelect,", ",sAlias[i],".",hdAttr," as ",hdAttr,"_",i,", ",sAlias[i],".recordid as recordid_",DT[i])
sCreateQuery<-paste0(sCreateQuery,", ",hdAttr,"_",i," int, recordid_",DT[i]," int")
if (i==1) sCreateQuery<-paste0(sCreateQuery,') PARTITIONED BY (scf string) STORED AS ORC')
if (i==1) sSelect<-paste0(sSelect,", a.scf")
if (i!=nMonths) sFrom<-paste0(sFrom," inner join historicdata.",jointabs[i]," ",sAlias[i]," on ",
paste(paste0(paste0("a.",c("scf","idkey01","ssn7")),"=",
paste0(sAlias[i],".",c("scf","idkey01","ssn7"))),collapse=" AND "))
}
system(paste0('beeline -u "jdbc:hive2://myserver1.com,myserver2.com,myserver3.com,myserver4.com,myserver5.com/work;\
serviceDiscoveryMode=zooKeeper;zooKeeperNamespace=hiveserver2" -e "',sCreateQuery,'"'))
system(paste0("beeline -u \"jdbc:hive2://myserver1.com,myserver2.com,myserver3.com,myserver4.com,myserver5.com/work;\
serviceDiscoveryMode=zooKeeper;zooKeeperNamespace=hiveserver2\" -e \"",sSelect," ",sFrom,"\""))