R語言讀取Hive數據表

R通過RJDBC包連接Hive

目前Hive集群是可以通過跳板機來訪問 HiveServer， 將Hive 中的批量數據讀入R環境，並進行后續的模型和算法運算。

1. 登錄跳板機后需要首先在Linux下進行配置

從root權限切換到bzsys用戶，才能夠通過驗證讀取Hive。首先需要配置Hadoop的臨時環境變量CLASSPATH路徑

su bzsys

export CLASSPATH= $CLASSPATH :/etc/hadoop/conf

2. 下載並安裝RJDBC包

Linux下直接運行R的install 函數通常不成功，建議下載預編譯的包通過命令行安裝，如下載文件 RJDBC_0.2-6.tar.gz (http://www.rforge.net/RJDBC/)

切換到下載包所在的文件夾下，如 /etc/usr/R-patched/packages (替換為自己的目錄)，在終端運行R的命令行，完成安裝。

R CMD INSTALL RJDBC_0.2-6.tar.gz

3. 進入R環境

library(RJDBC) # 載入RJDBC包

# 設置R連接時類的路徑 CLASSPATH，注意一定要引hive/hadoop這三個路徑下的所有包才可以通過認證

cp = c(list.files( "/usr/lib/hive/lib" , pattern = "[.]jar" , full.names=TRUE, recursive=TRUE),

list.files( "/usr/lib/hadoop" , pattern = "[.]jar" , full.names=TRUE, recursive=TRUE),

list.files( "/etc/hadoop/conf" , full.names=TRUE, recursive=TRUE),

recursive=TRUE

)

# 新建RJDBC的 Driver

drv <- JDBC(driverClass = "org.apache.hive.jdbc.HiveDriver" , classPath = cp )

# 建立連接Connection

# 常見錯誤: 注意路徑是jdbc:hive2:而不是jdbc:hive:, 因為新版Hive啟用了 HiveServer2，替代了之前的HiveServer變量

# "hadoop-jy-backupserver:10000" 為濟陽集群的URL和默認端口PORT 10000, "principal=" 為需要的特殊認證Authentication

hiveconnection <- dbConnect(drv, "jdbc:hive2://hadoop-jy-backupserver:10000/default;principal=hive/[e-mail]" ,user= "*******" , password= "*******" )   # *** 替換為相應用戶名和密碼

4. R操作 Hive數據表范例

library(RJDBC)

cp = c(list.files( "/usr/lib/hive/lib" , pattern = "[.]jar" , full.names=TRUE, recursive=TRUE),

list.files( "/usr/lib/hadoop" , pattern = "[.]jar" , full.names=TRUE, recursive=TRUE),

list.files( "/etc/hadoop/conf" , full.names=TRUE, recursive=TRUE),

recursive=TRUE

)

drv = JDBC(driverClass = "org.apache.hive.jdbc.HiveDriver" , classPath = cp )

hiveconnection = dbConnect(drv, "jdbc:hive2://hadoop-jy-backupserver:10000/default;principal=hive/[e-mail]" ,user= "*******" , password= "*******" )   # *** 替換為相應用戶名密碼

 

# 從BAIDU_INDEX 表中查詢記錄的個數，存入DataFrame

count = dbGetQuery(hiveconnection, "SELECT count(*) FROM cpr.baidu_index" )  # 開始執行MapReduce任務

 

# 非查詢Query的語句,如創建表CREATE, dbSendUpdate 函數執行所有非查詢Query的語句

sqlCreateTbl = "CREATE TABLE IF NOT EXISTS cpr.person_correlation_graph_temp(rownames STRING,ibao_person_id_x STRING, ibao_person_id_y STRING,

     cor_index DOUBLE,start_date STRING,end_date STRING)"

result=dbSendUpdate(hiveconnection,sqlCreateTbl)

 

# 將數據寫入Hive數據庫

dfToLoad = data.frame(rownames=c( '1' , '2' ),ibao_person_id_x=c( '盜墓筆記' , '盜墓筆記' ),

ibao_person_id_y=c( '李易峰' , '楊洋' ),cor_index=c(0.8900,0.5100))

dbWriteTable(hiveconnection, "cpr.person_correlation_graph" , dfToLoad, overwrite=TRUE)

 

# R中執行其他Hive SQL的函數

dbListTables(hiveconnection, "%qiyu%" )

df = dbReadTable(hiveConn, "iris" )