用java代碼調用shell腳本執行sqoop將hive表中數據導出到mysql

1：創建shell腳本 

touch sqoop_options.sh
chmod 777 sqoop_options.sh

編輯文件  特地將執行map的個數設置為變量  測試 可以java代碼傳參數 同時也驗證sqoop的 options 屬性支持這種寫法

#!/bin/bash
/opt/cdh-5.3.6/sqoop-1.4.5-cdh5.3.6/bin/sqoop --options-file /opt/cdh-5.3.6/sqoop-1.4.5-cdh5.3.6/sqoop-import-mysql.txt --num-mappers $1
if [ $? -eq 0 ];then
 echo "success"
else
 echo "error"
fi

2：創建  sqoop-import-mysql.txt 文件並編輯

touch sqoop-import-mysql.txt

export
--connect
jdbc:mysql://172.16.71.27:3306/babasport
--username
root
--password
root
--table
test_hive
--export-dir
/user/hive/warehouse/hive_bbs_product_snappy
--input-fields-terminated-by
'\t'

 

hive數據存在hdfs位置

 

3：開始寫java后台代碼   目前只支持 window寫法 后期加上linux調用shell腳本的寫法

package com.liveyc.common.utils;

import java.util.Properties;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

public  class FileToHbase {  
    /**
     * shell腳本執行成功標識
     */
    public static int SHELL_EXIT_OK = 0;
    public static Log log = LogFactory.getLog(FileToHbase.class);
    public static String connIp = "172.16.71.120";
    public static String connUser = "root";
    public static String connPwd = "123456";
    
    public static void main(String[] args) throws Exception {
        boolean result = export();
        System.out.println(result);
    }
    
    public static boolean export() throws Exception {
        boolean result = false;
        // 如果當前系統是window系統需要遠程ssh連接系統
        if (isWinSystem()) {
            ConnectShell connectShell = new ConnectShell(connIp, connUser, connPwd, "utf-8");
            String url = "/opt/cdh-5.3.6/sqoop-1.4.5-cdh5.3.6/sqoop_options.sh" + " " +1;
            result = connectShell.excuteShellCommand(url);
        } 
        return result;
    }

    /**
     * 當前操作系統類型
     * 
     * @return true 為windos系統，false為linux系統
     */
    public static boolean isWinSystem() {
        // 獲取當前操作系統類型
        Properties prop = System.getProperties();
        String os = prop.getProperty("os.name");
        if (os.startsWith("win") || os.startsWith("Win")) {
            return true;
        } else {
            return false;
        }
    }
}  

 

package com.liveyc.common.utils;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.UnsupportedEncodingException;
import java.nio.charset.Charset;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import ch.ethz.ssh2.ChannelCondition;
import ch.ethz.ssh2.Connection;
import ch.ethz.ssh2.Session;
import ch.ethz.ssh2.StreamGobbler;

/**
 * 
 * ConnectShell
 * 
 * @Description:連接Shell腳本所在服務器
 * @author:aitf
 * @date: 2016年3月31日
 *
 */
public class ConnectShell {
    private Connection conn;
    private String ipAddr;
    private String userName;
    private String password;
    private String charset = Charset.defaultCharset().toString();
    private static final int TIME_OUT = 1000 * 5 * 60;
    public static Log log = LogFactory.getLog(ConnectShell.class);

    public ConnectShell(String ipAddr, String userName, String password, String charset) {
        this.ipAddr = ipAddr;
        this.userName = userName;
        this.password = password;
        if (charset != null) {
            this.charset = charset;
        }
    }

    public boolean login() throws IOException {
        conn = new Connection(ipAddr);
        conn.connect();
        return conn.authenticateWithPassword(userName, password); // 認證
    }

    /**
     * 
     * @Title: excuteShellCommand
     * @Description: 執行shell腳本命令
     * @param shellpath
     * @return
     */
    public boolean excuteShellCommand(String shellpath) {
        InputStream in = null;
        boolean result = false;
        String str = "";
        try {
            if (this.login()) {
                Session session = conn.openSession();
                 //session.execCommand("cd /root");
                session.execCommand(shellpath);
                in = new StreamGobbler(session.getStdout());
                // in = session.getStdout();
                 str = this.processStdout(in, charset);
                session.waitForCondition(ChannelCondition.EXIT_STATUS, TIME_OUT);
                session.close();
                conn.close();  
                if (str.contains("success")) {
                    result = true;
                }else{
                    result = false;
                }
            } 
        } catch (IOException e1) {
            e1.printStackTrace();
        }
        return result;
    }

    public String excuteShellCommand2(String shellpath) throws Exception {
        InputStream in = null;
        String result = "";
        try {
            if (this.login()) {
                Process exec = Runtime.getRuntime().exec(shellpath);// ipconfig
                in = exec.getInputStream();
                result = this.processStdout(in, this.charset);
            }
        } catch (IOException e1) {
            e1.printStackTrace();
        }
        return result;
    }

    /**
     * 轉化結果
     * 
     * @param in
     * @param charset
     * @return
     * @throws UnsupportedEncodingException
     */
    public String processStdout(InputStream in, String charset) throws UnsupportedEncodingException {
        String line = null;
        BufferedReader brs = new BufferedReader(new InputStreamReader(in, charset));
        StringBuffer sb = new StringBuffer();
        try {
            while ((line = brs.readLine()) != null) {
                sb.append(line + "\n");
            }
        } catch (IOException e) {
            log.error("---轉化出現異常---");
        }
        return sb.toString();
    }

}

 

4：開始測試

在mysql創建一個表  hive中數據格式 是  int int String

CREATE TABLE test_hive(
id INT,
brand_id INT,
NAME VARCHAR(200)
)

 

 執行java main方法 開始測試

觀看8088端口 查看MapReduce的運行狀況 發現正在運行（開心）

 

 執行完畢  

 

 可以看到 只有1個 MapReduce任務 （默認的個數是4個 這樣看來第一步寫的shell腳本 參數是傳遞過來了 sqoop的 options 也支持這種直接指定參數的寫法）

現在轉過來看java代碼

返回值 ：

Warning: /opt/cdh-5.3.6/sqoop-1.4.5-cdh5.3.6/bin/../../hbase does not exist! HBase imports will fail.
Please set $HBASE_HOME to the root of your HBase installation.
Warning: /opt/cdh-5.3.6/sqoop-1.4.5-cdh5.3.6/bin/../../hcatalog does not exist! HCatalog jobs will fail.
Please set $HCAT_HOME to the root of your HCatalog installation.
Warning: /opt/cdh-5.3.6/sqoop-1.4.5-cdh5.3.6/bin/../../accumulo does not exist! Accumulo imports will fail.
Please set $ACCUMULO_HOME to the root of your Accumulo installation.
Warning: /opt/cdh-5.3.6/sqoop-1.4.5-cdh5.3.6/bin/../../zookeeper does not exist! Accumulo imports will fail.
Please set $ZOOKEEPER_HOME to the root of your Zookeeper installation.
success

 

 

 發現返回 success 說明shell腳本執行成功了  

 

一切執行正常   看下mysql 數據庫表中有沒有數據

 

 

 

 OK 一切正常 ， 后期把linux執行shell腳本的語句也補充上 。