通过Python在Hive中实现层次查询

INSERT INTO `emp` VALUES ('7369', 'SMITH', 'CLERK', '7902', '1980-12-17', 800.00, NULL, 20);INSERT INTO `emp` VALUES ('7499', 'ALLEN', 'SALESMAN', '7698', '1981-2-20', 1600.00, 300.00, 30);INSERT INTO `emp` VALUES ('7521', 'WARD', 'SALESMAN', '7698', '1981-2-22', 1250.00, 500.00, 30);INSERT INTO `emp` VALUES ('7566', 'JONES', 'MANAGER', '7839', '1981-4-2', 2975.00, NULL, 20);INSERT INTO `emp` VALUES ('7654', 'MARTIN', 'SALESMAN', '7698', '1981-9-28', 1250.00, 1400.00, 30);INSERT INTO `emp` VALUES ('7698', 'BLAKE', 'MANAGER', '7839', '1981-5-1', 2850.00, NULL, 30);INSERT INTO `emp` VALUES ('7782', 'CLARK', 'MANAGER', '7839', '1981-6-9', 2450.00, NULL, 10);INSERT INTO `emp` VALUES ('7788', 'SCOTT', 'ANALYST', '7566', '1987-7-13', 3000.00, NULL, 20);INSERT INTO `emp` VALUES ('7839', 'KING', 'PRESIDENT', NULL, '1981-11-17', 5000.00, NULL, 10);INSERT INTO `emp` VALUES ('7844', 'TURNER', 'SALESMAN', '7698', '1981-9-8', 1500.00, 0.00, 30);INSERT INTO `emp` VALUES ('7900', 'JAMES', 'CLERK', '7698', '1981-12-3', 950.00, NULL, 30);INSERT INTO `emp` VALUES ('7902', 'FORD', 'ANALYST', '7566', '1981-12-3', 3000.00, NULL, 20);INSERT INTO `emp` VALUES ('7876', 'ADAMS', 'CLERK', '7788', '1987-7-13', 1100.00, NULL, 20);

WITH RECURSIVE TEMP(EMPNO, ENAME, MGR, LEVEL, PATH) AS(SELECT EMPNO, ENAME, MGR, 1 , CAST(ENAME AS CHAR(1000)) FROM EMP WHERE MGR IS NULLUNION ALLSELECT T2.EMPNO, T2.ENAME, T2.MGR, T1.LEVEL+1, CONCAT(T1.PATH, '->', T2.ENAME)FROM TEMP T1    ,EMP T2WHERE T1.EMPNO = T2.MGR)SELECT * FROM TEMP

create table emp(  `empno` string,   `ename` string,   `job` string,   `mgr` string,   `hiredate` string,   `sal` string,   `comm` string,   `deptno` string)row format delimited fields terminated by ','

load data local inpath '/root/emp.txt' into table emp

#_*_encoding:utf-8_*_import copyimport sysdef get_data(sep, root_empno):    data = []   # 存储所有数据    first_level_data = []  # 第一层的数据    for line in sys.stdin:        line = line.strip()        temp_data = line.split(sep)        # empno,ename,job,mgr,hiredate,sal,comm,deptno        empno = temp_data[0]        ename = temp_data[1]        mgr = temp_data[2]        if root_empno == empno:            first_level_data.append([empno, ename, mgr, 1, ename])  # 最后两个 level, path        data.append(temp_data)    return first_level_data, datadef handle_data(sep, first_level_data, data):    final_data = copy.deepcopy(first_level_data)    total_cnt = len(data)    for current_level in range(2, total_cnt + 1): # 逐层处理        last_level = current_level - 1  # 上一层 等于当前层减法1        loop_flag = 0        for item in final_data:            empno = item[0]            ename = item[1]            mgr = item[2]            level = item[3]            ename_path = item[4]            if last_level != level:                continue            else:                loop_flag = 1                for d in data:                    temp_list = []                    if empno == d[2]: #上一层的empno  是 当前的层的mgr                        temp_list.extend(d)                        temp_list.append(current_level)                        temp_list.append(ename_path + "->" + d[1])                        final_data.append(temp_list)        if loop_flag == 0:            break    final_data.sort(key=lambda d: d[3])    for i in final_data:        d = [str(j) for j in i]        print(sep.join(d))def main():    sep = ','    root_empno = sys.argv[1]   # 根节点的 员工编号    first_level_data, data = get_data(sep, root_empno)    handle_data(sep, first_level_data, data)if __name__ == '__main__':    main()

set mapreduce.framework.name = local;add file file:////root/hive_py/cte.py;selecttransform(empno, ename, mgr) row format delimitedfields terminated by ','using 'python cte.py 7839' as (k1,k2,k3,k4,k5 )row format delimitedfields terminated by ','from emp; delete file file:////root/hive_py/cte.py;