0

我创建了可以从 Hive CLI 和/或beeline CLI 和/或 Spark (2.3.1) WITH 运行的 select 和 join 语句enableHiveSupport=TRUE。（注意：我的 API 使用 SparkR）

使用直线连接和写入需要 30 分钟，但使用 Spark 连接和写入enableHiveSupport=TRUE需要 3.5 小时。这要么意味着 Spark 和它的连接器是垃圾，要么我没有按照我应该的方式使用 spark……而且我读到的关于 Spark 的“自切片面包以来最好的东西”评论的所有内容都意味着我可能没有正确使用它。

我想从 Hive 表中读取，但我不希望 Hive 做任何事情。我想对每月数据运行连接，对每条记录的每月增量运行回归，然后将我的最终斜率/beta 输出到实木复合地板中的输出表，如果需要，可以从 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,"\""))
4

0 回答 0