快捷導(dǎo)航

Java API如何實(shí)現(xiàn)向Hive批量導(dǎo)入數(shù)據(jù)

更新時(shí)間：2021年07月22日 10:45:37 作者：盛裝吾步

這篇文章主要介紹了Java API如何實(shí)現(xiàn)向Hive批量導(dǎo)入數(shù)據(jù)的操作，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教

Java API實(shí)現(xiàn)向Hive批量導(dǎo)入數(shù)據(jù)

Java程序中產(chǎn)生的數(shù)據(jù)，如果導(dǎo)入oracle或者mysql庫(kù)，可以通過(guò)jdbc連接insert批量操作完成，但是當(dāng)前版本的hive并不支持批量insert操作，因?yàn)樾枰葘⒔Y(jié)果數(shù)據(jù)寫(xiě)入hdfs文件，然后插入Hive表中。

package com.enn.idcard; 
import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.List;
 
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
/**
 * <p>Description: </p>
 * @author kangkaia
 * @date 2017年12月26日 下午1:42:24
 */
public class HiveJdbc { 
    public static void main(String[] args) throws IOException {
    	List<List> argList = new ArrayList<List>();
		List<String> arg = new ArrayList<String>();
		arg.add("12345");
		arg.add("m");
		argList.add(arg);
		arg = new ArrayList<String>();
		arg.add("54321");
		arg.add("f");
		argList.add(arg);
//		System.out.println(argList.toString());
		String dst = "/test/kk.txt";
		createFile(dst,argList);
		loadData2Hive(dst);
    }
 
    /**
     * 將數(shù)據(jù)插入hdfs中，用于load到hive表中，默認(rèn)分隔符是"\001"
     * @param dst
     * @param contents
     * @throws IOException
     */
    public static void createFile(String dst , List<List> argList) throws IOException{
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(conf);
        Path dstPath = new Path(dst); //目標(biāo)路徑
        //打開(kāi)一個(gè)輸出流
        FSDataOutputStream outputStream = fs.create(dstPath);
        StringBuffer sb = new StringBuffer();
        for(List<String> arg:argList){
			for(String value:arg){
				sb.append(value).append("\001");
			}
			sb.deleteCharAt(sb.length() - 4);//去掉最后一個(gè)分隔符
			sb.append("\n");
		}
        sb.deleteCharAt(sb.length() - 2);//去掉最后一個(gè)換行符
        byte[] contents =  sb.toString().getBytes();
        outputStream.write(contents);
        outputStream.close();
        fs.close();
        System.out.println("文件創(chuàng)建成功！");        
    }
    /**
     * 將HDFS文件load到hive表中
     * @param dst
     */
    public static void loadData2Hive(String dst) {
    	String JDBC_DRIVER = "org.apache.hive.jdbc.HiveDriver";
    	String CONNECTION_URL = "jdbc:hive2://server-13:10000/default;auth=noSasl";
    	String username = "admin";
        String password = "admin";
        Connection con = null;
		
		try {
			Class.forName(JDBC_DRIVER);
			con = (Connection) DriverManager.getConnection(CONNECTION_URL,username,password);
			Statement stmt = con.createStatement();			
			String sql = " load data inpath '"+dst+"' into table population.population_information ";
			
			stmt.execute(sql);
			System.out.println("loadData到Hive表成功！");
		} catch (SQLException e) {
			e.printStackTrace();
		} catch (ClassNotFoundException e) {
			e.printStackTrace();
		}finally {
			// 關(guān)閉rs、ps和con
			if(con != null){
				try {
					con.close();
				} catch (SQLException e) {
					e.printStackTrace();
				}
			}
		}
	}    
}

注意：

本例使用mvn搭建，conf配置文件放在src/main/resources目錄下。

Hive提供的默認(rèn)文件存儲(chǔ)格式有textfile、sequencefile、rcfile等。用戶也可以通過(guò)實(shí)現(xiàn)接口來(lái)自定義輸入輸?shù)奈募袷健?/p>

在實(shí)際應(yīng)用中，textfile由于無(wú)壓縮，磁盤及解析的開(kāi)銷都很大，一般很少使用。Sequencefile以鍵值對(duì)的形式存儲(chǔ)的二進(jìn)制的格式，其支持針對(duì)記錄級(jí)別和塊級(jí)別的壓縮。rcfile是一種行列結(jié)合的存儲(chǔ)方式（text file和sequencefile都是行表[row table]），其保證同一條記錄在同一個(gè)hdfs塊中，塊以列式存儲(chǔ)。一般而言，對(duì)于OLTP而言，行表優(yōu)勢(shì)大于列表，對(duì)于OLAP而言，列表的優(yōu)勢(shì)大于行表，特別容易想到當(dāng)做聚合操作時(shí)，列表的復(fù)雜度將會(huì)比行表小的多，雖然單獨(dú)rcfile的列運(yùn)算不一定總是存在的，但是rcfile的高壓縮率確實(shí)減少文件大小，因此實(shí)際應(yīng)用中，rcfile總是成為不二的選擇，達(dá)觀數(shù)據(jù)平臺(tái)在選擇文件存儲(chǔ)格式時(shí)也大量選擇了rcfile方案。

通過(guò)hdfs導(dǎo)入hive的表默認(rèn)是textfile格式的，因此可以改變存儲(chǔ)格式，具體方法是先創(chuàng)建sequencefile、rcfile等格式的空表，然后重新插入數(shù)據(jù)即可。

insert overwrite table seqfile_table select * from textfile_table; 
……
insert overwrite table rcfile_table select * from textfile_table;

java 批量插入hive中轉(zhuǎn)在HDFS

稍微修改了下，這文章是通過(guò)將數(shù)據(jù)存盤后，加載到HIVE.

模擬數(shù)據(jù)放到HDFS然后加載到HIVE，請(qǐng)大家記得添加HIVE JDBC依賴否則會(huì)報(bào)錯(cuò)。

加載前的數(shù)據(jù)表最好用外部表，否則會(huì)drop表的時(shí)候元數(shù)據(jù)會(huì)一起刪除！

  <dependency>
   <groupId>org.apache.hive</groupId>
   <artifactId>hive-jdbc</artifactId>
   <version>1.1.0</version>
  </dependency>

代碼

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.List;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
public class Demo {
	    public static void main(String[] args) throws Exception {
	    	List<List> argList = new ArrayList<List>();
			List<String> arg = new ArrayList<String>();
			arg.add("12345");
			arg.add("m");
			argList.add(arg);
			arg = new ArrayList<String>();
			arg.add("54321");
			arg.add("f");
			argList.add(arg);
//			System.out.println(argList.toString());
			String dst = "/test/kk.txt";
			createFile(dst,argList);
//			loadData2Hive(dst);
	    }
	    /**
	     * 將數(shù)據(jù)插入hdfs中，用于load到hive表中，默認(rèn)分隔符是"|"
	     * @param dst
	     * @param contents
	     * @throws IOException
	     * @throws Exception 
	     * @throws InterruptedException 
	     */
	    public static void createFile(String dst , List<List> argList) throws IOException, InterruptedException, Exception{
	        Configuration conf = new Configuration();
			FileSystem fs = FileSystem.get(new URI("hdfs://hadoop:9000"),conf,"root");
	        Path dstPath = new Path(dst); //目標(biāo)路徑
	        //打開(kāi)一個(gè)輸出流
	        FSDataOutputStream outputStream = fs.create(dstPath);
	        StringBuffer sb = new StringBuffer();
	        for(List<String> arg:argList){
				for(String value:arg){
					sb.append(value).append("|");
				}
				sb.deleteCharAt(sb.length() - 1);//去掉最后一個(gè)分隔符
				sb.append("\n");
			}
	        byte[] contents =  sb.toString().getBytes();
	        outputStream.write(contents);
			outputStream.flush();;
	        outputStream.close();
	        fs.close();
	        System.out.println("文件創(chuàng)建成功！");
	        
	    }
	    /**
	     * 將HDFS文件load到hive表中
	     * @param dst
	     */
	    public static void loadData2Hive(String dst) {
	    	String JDBC_DRIVER = "org.apache.hive.jdbc.HiveDriver";
	    	String CONNECTION_URL = "jdbc:hive2://hadoop:10000/default";
	    	String username = "root";
	        String password = "root";
	        Connection con = null;
			
			try {
				Class.forName(JDBC_DRIVER);
				con = (Connection) DriverManager.getConnection(CONNECTION_URL,username,password);
				Statement stmt = con.createStatement();
				
				String sql = " load data inpath '"+dst+"' into table test ";//test 為插入的表
				
				stmt.execute(sql);
				System.out.println("loadData到Hive表成功！");
			} catch (SQLException e) {
				e.printStackTrace();
			} catch (ClassNotFoundException e) {
				e.printStackTrace();
			}finally {
				// 關(guān)閉rs、ps和con
				if(con != null){
					try {
						con.close();
					} catch (SQLException e) {
						e.printStackTrace();
					}
				}
			}
		}
	    
	}

以上為個(gè)人經(jīng)驗(yàn)，希望能給大家一個(gè)參考，也希望大家多多支持腳本之家。

您可能感興趣的文章: