SpringBoot連接Hive實(shí)現(xiàn)自助取數(shù)

Hive系列文章

  1. Hive表的基本操作
  2. Hive中的集合數(shù)據(jù)類型
  3. Hive動(dòng)態(tài)分區(qū)詳解
  4. hive中orc格式表的數(shù)據(jù)導(dǎo)入
  5. Java通過(guò)jdbc連接hive
  6. 通過(guò)HiveServer2訪問Hive
  7. SpringBoot連接Hive實(shí)現(xiàn)自助取數(shù)
  8. hive關(guān)聯(lián)hbase表
  9. Hive udf 使用方法
  10. Hive基于UDF進(jìn)行文本分詞
  11. Hive窗口函數(shù)row number的用法
  12. 數(shù)據(jù)倉(cāng)庫(kù)之拉鏈表

公司運(yùn)營(yíng)免不了讓我們數(shù)據(jù)做一些臨時(shí)取數(shù)，這些取數(shù)有時(shí)候是重復(fù)的，或者可以做成可配置的。需要開發(fā)成界面，供他們選擇，自然想到SpringBoot連接Hive，可以把取數(shù)做成一鍵生成，或者讓他們自己寫sql，通常大多人是不會(huì)sql的。

1. 需要的依賴配置

為了節(jié)省篇幅，這里給出hiveserver2方式連接hive主要的maven依賴，父工程springboot依賴省略。

<!-- 版本信息 -->
<properties>
    <hadoop.version>2.6.5</hadoop.version>
    <mybatis.version>3.2.7</mybatis.version>
    <scopeType>compile</scopeType>
</properties>
<dependency>
    <groupId>org.mybatis</groupId>
    <artifactId>mybatis</artifactId>
    <version>{mybatis.version}</version>
</dependency>

<!-- hadoop依賴 -->
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-common</artifactId>
    <version>{hadoop.version}</version>
    <scope>{scopeType}</scope>
</dependency>

<!-- hive-jdbc -->
<!-- https://mvnrepository.com/artifact/org.apache.hive/hive-jdbc -->
<dependency>
    <groupId>org.apache.hive</groupId>
    <artifactId>hive-jdbc</artifactId>
    <exclusions>
        <exclusion>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
        </exclusion>
        <exclusion>
            <groupId>ch.qos.logback</groupId>
            <artifactId>logback-core</artifactId>
        </exclusion>
        <exclusion>
            <groupId>ch.qos.logback</groupId>
            <artifactId>logback-classic</artifactId>
        </exclusion>
    </exclusions>
    <version>1.2.1</version>
    <scope>{scopeType}</scope>
</dependency>

<!-- 解析html -->
<dependency>
    <groupId>org.jsoup</groupId>
    <artifactId>jsoup</artifactId>
    <version>1.8.3</version>
</dependency>
XML

application-test.yml配置數(shù)據(jù)庫(kù)連接，這里用的是druid連接池管理hiveserver2連接，也是沒有問題的。

# Spring配置
spring:
  datasource:
    type: com.alibaba.druid.pool.DruidDataSource
    driverClassName: com.mysql.cj.jdbc.Driver
    druid:
      # 多數(shù)據(jù)源**省略若干***
      # hive數(shù)據(jù)源
      slave3:
        # 從數(shù)據(jù)源開關(guān)/默認(rèn)關(guān)閉
        enabled: true
        driverClassName: org.apache.hive.jdbc.HiveDriver
        url: jdbc:hive2://cdh:10000/default
        username: bigdata
        password: bigdata

2. 代碼實(shí)現(xiàn)

代碼實(shí)現(xiàn)跟其它程序一樣，都是mapper、service、controller層，套路一模一樣。一共設(shè)置了實(shí)時(shí)和離線兩個(gè)yarn資源隊(duì)列，由于其它部門人使用可能存在隊(duì)列壓力過(guò)大的情況，需要對(duì)數(shù)據(jù)量按照每次查詢的數(shù)據(jù)范圍不超過(guò)60天來(lái)限制，和此時(shí)集群使用資源不能大于55%，這里重點(diǎn)說(shuō)明一下controller層對(duì)數(shù)據(jù)量的預(yù)防。

實(shí)體類UserModel：

@NoArgsConstructor
@AllArgsConstructor
@Data
@ToString
public class UserModel extends BaseEntity{

    private String userId;
    private Integer count;
}
Java

2.1 集群資源使用率不大于55%

因?yàn)楹芏鄻I(yè)務(wù)查詢邏輯controller都要用到數(shù)據(jù)量防御過(guò)大的問題，這里使用了被Spring切面關(guān)聯(lián)的注解來(lái)標(biāo)識(shí)controller。

定義切面YarnResourceAspect，并且關(guān)聯(lián)注解@YarnResource

@Target({ElementType.TYPE, ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
public @interface YarnResource {

}

@Aspect
@Component
public class YarnResourceAspect {

    private static final Logger log = LoggerFactory.getLogger(YarnResourceAspect.class);

    /**
     * 配置切入點(diǎn)
     */
    @Pointcut("@annotation(com.ruoyi.common.annotation.YarnResource)")
    public void yarnResourcdPointCut(){
    }

    /**
     * 檢查yarn的資源是否可用
     */
    @Before("yarnResourcdPointCut()")
    public void before(){
        log.info("************************************檢查yarn的資源是否可用*******************************");
        // yarn資源緊張
        if(!YarnClient.yarnResourceOk()){
            throw new InvalidStatusException();
        }
    }

}
Java






獲取yarn的資源使用數(shù)據(jù)：

@Slf4j
public class YarnClient {

    /**
     * yarn資源不能超過(guò)多少
     */
    private static final int YARN_RESOURCE = 55;

    /**
     *
     * @return true : 表示資源正常， false: 資源緊張
     */
    public static boolean yarnResourceOk() {
        try {
            URL url = new URL("http://master:8088/cluster/scheduler");
            HttpURLConnection conn = null;
            conn = (HttpURLConnection) url.openConnection();
            conn.setRequestMethod("GET");
            conn.setUseCaches(false);
            // 請(qǐng)求超時(shí)5秒
            conn.setConnectTimeout(5000);
            // 設(shè)置HTTP頭:
            conn.setRequestProperty("Accept", "*/*");
            conn.setRequestProperty("User-Agent", "Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/86.0.4240.111 Safari/537.36");
            // 連接并發(fā)送HTTP請(qǐng)求:
            conn.connect();

            // 判斷HTTP響應(yīng)是否200:
            if (conn.getResponseCode() != 200) {
                throw new RuntimeException("bad response");
            }
            // 獲取所有響應(yīng)Header:
            Map<String, List<String>> map = conn.getHeaderFields();
            for (String key : map.keySet()) {
                System.out.println(key + ": " + map.get(key));
            }
            // 獲取響應(yīng)內(nèi)容:
            InputStream input = conn.getInputStream();
            byte[] datas = null;

            try {
                // 從輸入流中讀取數(shù)據(jù)
                datas = readInputStream(input);
            } catch (Exception e) {
                e.printStackTrace();
            }
            String result = new String(datas, "UTF-8");// 將二進(jìn)制流轉(zhuǎn)為String

            Document document = Jsoup.parse(result);

            Elements elements = document.getElementsByClass("qstats");

            String[] ratios = elements.text().split("used");

            return Double.valueOf(ratios[3].replace("%", "")) < YARN_RESOURCE;
        } catch (IOException e) {
            log.error("yarn資源獲取失敗");
        }

        return false;

    }

    private static byte[] readInputStream(InputStream inStream) throws Exception {
        ByteArrayOutputStream outStream = new ByteArrayOutputStream();
        byte[] buffer = new byte[1024];
        int len = 0;
        while ((len = inStream.read(buffer)) != -1) {
            outStream.write(buffer, 0, len);
        }
        byte[] data = outStream.toByteArray();
        outStream.close();
        inStream.close();
        return data;
    }
}
Java

在controller上通過(guò)注解@YarnResource標(biāo)識(shí)：

@Controller
@RequestMapping("/hero/hive")
public class HiveController {

    /**
     * html 文件地址前綴
     */
    private String prefix = "hero";

    @Autowired
    IUserService iUserService;

    @RequestMapping("")
    @RequiresPermissions("hero:hive:view")
    public String heroHive(){
        return prefix + "/hive";
    }

    @YarnResource
    @RequestMapping("/user")
    @RequiresPermissions("hero:hive:user")
    @ResponseBody
    public TableDataInfo user(UserModel userModel){
        DateCheckUtils.checkInputDate(userModel);

        PageInfo pageInfo = iUserService.queryUser(userModel);
        TableDataInfo tableDataInfo = new TableDataInfo();

        tableDataInfo.setTotal(pageInfo.getTotal());
        tableDataInfo.setRows(pageInfo.getList());

        return tableDataInfo;
    }
}
Java

2.2 查詢數(shù)據(jù)跨度不超過(guò)60天檢查

這樣每次請(qǐng)求進(jìn)入controller的時(shí)候就會(huì)自動(dòng)檢查查詢的日期是否超過(guò)60天了，防止載入數(shù)據(jù)過(guò)多，引發(fā)其它任務(wù)資源不夠。

public class DateCheckUtils {

    /**
     * 對(duì)前臺(tái)傳入過(guò)來(lái)的日期進(jìn)行判斷，防止查詢大量數(shù)據(jù)，造成集群負(fù)載過(guò)大
     * @param o
     */
    public static void checkInputDate(BaseEntity o){
        if("".equals(o.getParams().get("beginTime")) && "".equals(o.getParams().get("endTime"))){
            throw new InvalidTaskException();
        }

        String beginTime = "2019-01-01";
        String endTime = DateUtils.getDate();

        if(!"".equals(o.getParams().get("beginTime"))){
            beginTime = String.valueOf(o.getParams().get("beginTime"));
        }

        if(!"".equals(o.getParams().get("endTime"))){
            endTime = String.valueOf(o.getParams().get("endTime"));
        }

        // 查詢數(shù)據(jù)時(shí)間跨度大于兩個(gè)月
        if(DateUtils.getDayBetween(beginTime, endTime) > 60){
            throw new InvalidTaskException();
        }
    }
}
Java

這里訪問hive肯定需要切換數(shù)據(jù)源的，因?yàn)槠渌?yè)面還有對(duì)mysql的數(shù)據(jù)訪問，需要注意一下。

目前功能看起來(lái)很簡(jiǎn)單，沒有用到什么高大上的東西，后面慢慢完善。


作者：柯廣的網(wǎng)絡(luò)日志

微信公眾號(hào)：Java大數(shù)據(jù)與數(shù)據(jù)倉(cāng)庫(kù)