Hive-SQL解析及应用

1
2
3
4
5
6

<dependency>
    <groupId>org.apache.hive</groupId>
    <artifactId>hive-exec</artifactId>
    <version>${hive.version}</version>
    <classifier>core</classifier>
</dependency>

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

import org.apache.hadoop.hive.ql.parse.ParseDriver;
   // 创建 ParseDriver
   ParseDriver pd = new ParseDriver();
   // 获得解析的 tree
   ASTNode tree = pd.parse(sql);
   if (LOGGER.isDebugEnabled()) {
     // 这里可以将 ASTree 打印出来, 用于调试
     // debug 可以用
     LOGGER.debug("start to analyze sql: {}, ASTNode: {}", sql, tree.dump());
   }
   while ((tree.getToken() == null) && (tree.getChildCount() > 0)) {
     tree = (ASTNode) tree.getChild(0);
   }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

// 这里初始化我们的 Visitor 逻辑
final NodeProcessor processor = new OurNodeProcessor();

// 规则
Map<Rule, NodeProcessor> rules = Maps.newLinkedHashMap();
Dispatcher disp = new DefaultRuleDispatcher(processor, rules, null);
GraphWalker ogw = new DefaultGraphWalker(disp);
final List<Node> topNodes = Lists.newArrayList(tree);

// 发车, 开始遍历
ogw.startWalking(topNodes, null);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

// OurProcessor.java 
@Override
  public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx, Object... nodeOutputs) throws SemanticException {
    ASTNode pt = (ASTNode) nd;

    switch (pt.getToken().getType()) {
      case HiveParser.TOK_TAB:
        // 这里是 insert 语句
        final String insertedTable = BaseSemanticAnalyzer.getUnescapedName((ASTNode) pt.getChild(0));
        break;
      case HiveParser.TOK_TABREF:
        // 这里是 FROM 语句后面的内容
        final String selectedTable = getTableName(pt);
        break;
      case HiveParser.TOK_DROPTABLE:
        // 这里是 DROP 语句
        final String droppedTable = getTableName(pt);
        break;
      case HiveParser.TOK_CREATETABLE:
        // 这里是 CREATE 语句
        final String createdTable = getTableName(pt);
        break;
      // 将上面获取到的各种类型的 Table Name 存储起来就 OK 了
  }
  return null;
}

1
2
3
4
5

WITH data1 AS (SELECT id, age FROM table1 WHERE id > 0),
data2 AS (SELECT id, name FROM table2 WHERE id > 0)

-- 通过 WITH 语句, 将复杂 SQL 拆解, 提高易读性
SELECT a.*, b.* FROM data1 JOIN data2 on a.id = b.id

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

-- test_job.sql
-- set mapreduce 参数
set hive.exec.max.dynamic.partitions.pernode=100000;
set hive.exec.dynamic.partition.mode=nostrick;
set hive.groupby.skewindata=true;

-- set 变量
set hivevar:ds=2016-01-01;
set hivevar:appid='app1';

-- 在临时库创建临时 Table, 存储结果
use tmpdb;
DROP TABLE IF EXISTS tmp_users;
CREATE TABBLE tmp_users AS SELECT a.*, b.* FROM userdb.users a JOIN userdb.user_data b on a.id = b.id WHERE a.update_time >= '${ds}' AND appid=${appid}

-- 切换到 prod db
USE etldb;
INSERT INTO daily_users SELECT * FROM tmp_users;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

{  
  "createdTables": [
    "tmpdb.tmp_users"
  ],
  "droppedTables": [
    "tmpdb.tmp_users"
  ],
  "inputTables": [
    "userdb.users",
    "userdb.user_data"
  ],
  "insertedTables": [
    "etldb.daily_users"
  ]
}