3小时吃掉HiveSQL：从基础语法到性能优化

--创建分区表
DROP TABLE IF EXISTS basedb.sales;
CREATE TABLE IF NOT EXISTS basedb.sales
(sale_id INT,
 product string COMMENT '产品名称',
 amount double COMMENT '销量')
 PARTITIONED BY (year INT, month INT)

-- 插入数据到指定分区(覆盖分区)
INSERT OVERWRITE  TABLE basedb.sales PARTITION (year='2025', month='01')
SELECT id, amount, date
FROM tmpdb.raw_sales;


-- 动态插入数据到分区(追加数据)
SET hive.exec.dynamic.partition = true;
SET hive.exec.dynamic.partition.mode = nonstrict;
INSERT INTO TABLE basedb.sales PARTITION (year, month)
SELECT id, amount, date, year, month
FROM tmpdb.raw_sales;

--数据查询select语句常用模版
SELECT ... 
    FROM 表1 JOIN 表2
ON 等值连接 
WHERE 分组前过滤条件 
GROUP BY 分组字段 
HAVING 分组后过滤条件 
ORDER BY 排序字段

-- 授权用户访问表权限
GRANTSELECT, INSERT, UPDATE, DELETEONTABLE basedb.sales TOUSER user1;
-- 撤销用户访问表权限
REVOKESELECT, INSERT, UPDATE, DELETEONTABLE basedb.sales FROMUSER user1;

-- 提交事务
COMMIT;
-- 回滚事务
ROLLBACK;

--创建同时有分区和分桶的表
CREATE TABLE orders (
    order_id INT,
    user_id INT,
    amount DOUBLE,
    order_date STRING
)
PARTITIONED BY (order_month STRING)  -- 按月份分区
CLUSTERED BY (user_id) INTO 10 BUCKETS  -- 按用户ID分桶，共10个桶
STORED AS ORC;  -- 使用ORC存储格式，提升查询性能

-- BucketMapJoin 支持大表对大表的连接，且不用shuffle.

--1，创建桶表
CREATE TABLE large_table_A (
    id INT,
    value STRING
)
CLUSTERED BY (id) INTO8 BUCKETS
STORED AS ORC;

CREATE TABLE large_table_B (
    id INT,
    description STRING
)
CLUSTERED BY (id) INTO8 BUCKETS
STORED AS ORC;

INSERTover large_table_A 


--2，启用 BucketMapJoin 优化
SET hive.optimize.bucketmapjoin=true;
SET hive.optimize.bucketmapjoin.sortedmerge=true;

-- 基于桶的连接查询
SELECT a.id, a.value, b.description
FROM large_table_A a
JOIN large_table_B b
ON a.id = b.id;

-- 隐式将 TINYINT 转为 INT
SELECT CAST(1 AS TINYINT) + 100;  

-- STRING 自动转为 DOUBLE
SELECT '3.14' + 2.0;

-- 数值转字符串
SELECT CAST(123.45 AS STRING);

-- 字符串转日期
SELECT CAST('2023-01-01' AS DATE);

-- 字符串转整数
SELECT CAST('100' AS INT);

SELECT DISTINCT column1, column2
FROM table1
JOIN table2 ON table1.id = table2.id
WHERE condition
GROUP BY column1
HAVING condition
ORDER BY column1
LIMIT 10;

--嵌套的子查询写法
SELECT subquery.column1, subquery.column2, table2.column3
FROM (
    SELECT column1, column2
    FROM table1
    WHERE condition
) AS subquery
JOIN table2 ON subquery.id = table2.id
WHERE another_condition
ORDER BY subquery.column1;

-- with  as 子查询写法
WITH subquery_name AS (
    SELECT column1, column2
    FROM table1
    WHERE condition
) --注意这里没有分号，with as 不是一条独立的语句，而是作为一个子查询
SELECT subquery_name.column1, subquery_name.column2, table2.column3
FROM subquery_name
JOIN table2 ON subquery_name.id = table2.id
WHERE another_condition
ORDER BY subquery_name.column1;

SELECT a.*, b.*
FROM table_a a
JOIN table_b b
  ON a.id = b.id;

SELECT a.*, b.*
FROM table_a a
LEFT OUTER JOIN table_b b
  ON a.id = b.id;

SELECT a.*, b.*
FROM table_a a
RIGHT OUTER JOIN table_b b
  ON a.id = b.id;

SELECT a.*, b.*
FROM table_a a
FULL OUTER JOIN table_b b
  ON a.id = b.id;

SELECT a.*
FROM table_a a
LEFT SEMI JOIN table_b b
  ON a.id = b.id;

--存在左表中，但是不在右表中
SELECT a.*
FROM table_a a
LEFT ANTI JOIN table_b b
  ON a.join_key = b.join_key;

-- 通过 LEFT JOIN 后判断右表字段是否为 NULL 来达到 anti join 的效果

SELECT a.*
FROM table_a a
LEFT JOIN table_b b
  ON a.join_key = b.join_key
WHERE b.join_key IS NULL;

SELECT a.*, b.*
FROM table_a a
CROSS JOIN table_b b;

SELECT column_a FROM table1
UNION ALL
SELECT column_a FROM table2;

SELECT DISTINCT column_a FROM (
  SELECT column_a FROM table1
  UNION ALL
  SELECT column_a FROM table2
) t;

DROP TABLE IF EXISTS base_db.predict_data_version;
CREATE TABLE IF NOTEXISTS base_db.predict_data_version AS
select'V1.1'as predict_version, 0as used_version unionall
select'V1.5'as predict_version, 0as used_version unionall
select'V2.0'as predict_version, 0as used_version unionall
select'V2.1.0'as predict_version, 0as used_version unionall
select'V2.1.5'as predict_version, 0as used_version unionall
select'V3.0.0'as predict_version, 1as used_version; 
  --线上应用数据数据

SELECT
  region,
  category,
COUNT(*) AS num_sales,
SUM(sales_amount) AS total_sales
FROM sales_data
GROUPBY region, category
HAVINGCOUNT(*) >10ANDSUM(sales_amount) >8000;

-- 各个版本不同日期销量预测总量对比
WITH summed_predictions AS (
select predict_version, cal_dt, sum(num_pred) as num_pred_sum
from mkt_db.sales_predict_data 
where cal_dt  between'2024-12-05'and'2025-01-22'
and predict_version in  ('V2.0','V3.0.0','V3.1.0')  
groupby predict_version, cal_dt
)
select cal_dt, 
SUM(CASEWHEN predict_version ='V2.0'THEN num_pred_sum ELSE0END) AS sum_V2_0,
SUM(CASEWHEN predict_version ='V3.0.0'THEN num_pred_sum ELSE0END) AS sum_V3_0_0,
SUM(CASEWHEN predict_version ='V3.1.0'THEN num_pred_sum ELSE0END) AS sum_V3_1_0
from summed_predictions
groupby cal_dt
orderby cal_dt;

<聚合/窗口函数> OVER (
    [PARTITION BY <分组列>]
    [ORDER BY <排序列>]
    [ROWS/RANGE BETWEEN <窗口范围>]
)

-- 未指定 窗口范围，计算的是全窗口内的总工资(每行都返回一样的值)
SELECT
    emp_id,
    salary,
    SUM(salary) OVER (
        PARTITIONBY department
    ) as window_sum
FROM
    employees;


-- 指定 ROWS，计算的是窗口范围内的累积工资(每行返回的值是递增的)
SELECT
    emp_id,
    salary,
    SUM(salary) OVER (
        PARTITIONBY department
        ORDERBY emp_id
        ROWSBETWEEN UNBOUNDED PRECEDING ANDCURRENTROW
    ) as cumulative_sum
FROM
    employees;

-- 1. 计算每个类别的销售排名：
SELECT
    category,
    sales_date,
    sales_amount,
    RANK() OVER (
        PARTITIONBY category 
        ORDERBY sales_amount DESC
    ) AS sales_rank
FROM sales_data;

-- 2. 计算每个类别的累计销售额
SELECT
    category,
    sales_date,
    sales_amount,
    SUM(sales_amount) OVER (
        PARTITIONBY category 
        ORDERBY sales_date
        ROWSBETWEEN UNBOUNDED PRECEDING ANDCURRENTROW
    ) AS cumulative_sales
FROM sales_data;


-- 3. 计算每个类别的移动平均销售额（计算前面6行和当前行的均值）
-- 如果 sales_date 列具有重复值（即同一天有多条记录），参与计算的依然是7行。

RANGEBETWEEN 是基于值的范围，所以它关注的是日期范围，而 ROWSBETWEEN 是基于行的范围，所以它关注的是行数。
SELECT
    category,
    sales_date,
    sales_amount,
    AVG(sales_amount) OVER (
        PARTITIONBY category
        ORDERBY sales_date
        ROWSBETWEEN6 PRECEDING ANDCURRENTROW
    ) AS rolling_avg
FROM sales_data;



-- 3. 计算每个类别的移动平均销售额（计算过去6天和当前行的均值）
-- 如果 sales_date 列具有重复值（即同一天有多条记录），参与计算的可能会超过7行。
SELECT
  category,
  sales_date,
  sales_amount,
AVG(sales_amount) OVER (
    PARTITIONBY category 
    ORDERBY sales_date 
    RANGEBETWEENINTERVAL6DAY PRECEDING ANDCURRENTROW
  ) AS rolling_avg
FROM
  sales_data;

-- (1) ROW_NUMBER(): 返回分区中每行的唯一行号，从1开始。
-- 假设salary取值： 8w, 7w, 5w, 5w, 3w, 1w, 1w
-- row_num结果取值：1,  2,  3,  4,  5,  6,  7
SELECT
    emp_id,
    salary,
    ROW_NUMBER() OVER (
        ORDERBY salary DESC
    ) as row_num
FROM
    employees;


-- (2) RANK(): 返回分区中每行的排名，相同值的行会有相同的排名，排名中有跳跃。
-- 假设salary取值： 8w, 7w, 5w, 5w, 3w, 1w, 1w
-- rank的 结果取值： 1,  2,  3,  3,  5,  6,  6
SELECT
    emp_id,
    salary,
    RANK() OVER (
        ORDERBY salary DESC
    ) as rank
FROM
    employees;

-- (3) DENSE_RANK(): 返回分区中每行的排名，相同值的行会有相同的排名，但排名不跳跃。
-- 假设salary取值： 8w, 7w, 5w, 5w, 3w, 1w, 1w
-- dense_rank取值： 1,  2,  3,  3,  4,  5,  5

SELECT
    emp_id,
    salary,
    DENSE_RANK() OVER (
        ORDERBY salary DESC
    ) as dense_rank
FROM
    employees;

-- (4) LAG(): 返回当前行前面第 N 行的值，若不存在则返回默认值。
--例子：计算销量前后值的差异
SELECT
  sales_date,
  sales_amount,
LAG(sales_amount, 1, 0) OVER (ORDERBY sales_date) AS previous_sales,
  sales_amount -LAG(sales_amount, 1, 0) OVER (ORDERBY sales_date) AS sales_difference
FROM
  sales_data;

-- (5) LEAD(): 返回当前行后面第 N 行的值，若不存在则返回默认值。
--例子：联合使用LEAD和LAG来计算滚动平均值(3日滚动)。
SELECT
  sales_date,
  sales_amount,
  (sales_amount +
   LAG(sales_amount, 1, 0) OVER (ORDERBY sales_date) +
   LEAD(sales_amount, 1, 0) OVER (ORDERBY sales_date)) /3AS moving_average
FROM
  sales_data;

-- 1，IF(condition, true_value, false_value)
SELECT
   order_id,
   amount,
   IF(amount >100, 'High', 'Low') AS amount_category
FROM orders;

-- 2，CASE WHEN condition THEN result [WHEN ...] ELSE result END
SELECT
   order_id,
   amount,
   CASE
      WHEN amount >100THEN'High'
      WHEN amount >50THEN'Medium'
      ELSE'Low'
   ENDAS amount_category
FROM orders;

-- 3，COALESCE(value1, value2, ..., valueN)
SELECT
   order_id,
   COALESCE(NULL, 'Default Value', amount) AS amount_value
FROM orders;

-- 4, NVL(value, default_value)
SELECT
   order_id,
   NVL(amount, 0) AS amount_value
FROM orders;

SELECT 
    CURRENT_DATE() as today,
    DATE_ADD(CURRENT_DATE(), 7) as next_week,
    DATE_FORMAT(CURRENT_TIMESTAMP(), 'yyyy-MM-dd HH:mm:ss') as formatted_time,
    DATEDIFF('2024-12-31', CURRENT_DATE()) as days_until_now;

-- 字符串长度
SELECT LENGTH('Hello') -- 返回5

-- 字符串连接
SELECT CONCAT('Hello', ' ', 'World') -- 返回'Hello World'

-- 字符串拼接(带分隔符)
SELECT CONCAT_WS(',', 'Hello', 'World') -- 返回'Hello,World'

-- 大小写转换
SELECTUPPER('hello') -- 返回'HELLO'
SELECTLOWER('HELLO') -- 返回'hello'

-- 子字符串
SELECT SUBSTR('Hello World', 1, 5) -- 返回'Hello'

-- 去除空格
SELECTTRIM('  Hello  ') -- 返回'Hello'
SELECT LTRIM('  Hello  ') -- 从左侧去除空格
SELECT RTRIM('  Hello  ') -- 从右侧去除空格

-- 查找子串位置
SELECT INSTR('Hello World', 'World') -- 返回7

-- 字符串替换
SELECT REGEXP_REPLACE('Hello World', 'World', 'Hive') -- 返回'Hello Hive'

-- 字符串分割
SELECT SPLIT('Hello,World,Hive', ',') -- 返回数组['Hello', 'World', 'Hive']

-- 获取字符串第一个出现位置
SELECT LOCATE('o', 'Hello World') -- 返回5

-- 重复字符串
SELECT REPEAT('Hi', 3) -- 返回'HiHiHi'

-- 字符串反转
SELECT REVERSE('Hello') -- 返回'olleH'

-- 从左/右填充字符
SELECT LPAD('Hi', 5, '*') -- 返回'***Hi'
SELECT RPAD('Hi', 5, '*') -- 返回'Hi***'

-- 统计每个部门的平均工资
SELECT dept_id, AVG(salary) 
FROM employees 
GROUP BY dept_id;

-- 谓词下推优化（将过滤条件下推到靠近源表）
SET hive.optimize.ppd=true;  -- 启用谓词下推
SET hive.optimize.ppd.storage=true;  -- 启用存储层谓词下推

-- 启用向量化
SET hive.vectorized.execution.enabled=true;
SET hive.vectorized.execution.reduce.enabled=true;

-- 启动并行执行 (每个worker多线程增加并行度)
SET hive.exec.parallel=true;
SET hive.exec.parallel.thread.number=16;  -- 线程数根据资源情况设置

-- 启动数据压缩 
SET hive.exec.compress.intermediate=true; --中间数据压缩
SET mapreduce.map.output.compress.codec=org.apache.hadoop.io.compress.SnappyCodec; --压缩格式
SET hive.exec.compress.output=true; --最终数据压缩
SET mapreduce.output.fileoutputformat.compress.codec=org.apache.hadoop.io.compress.SnappyCodec; --压缩格式

-- 合并小文件
SET hive.merge.mapfiles=true;
SET hive.merge.mapredfiles=true;
SET hive.merge.size.per.task=256000000;  -- 256MB
SET hive.merge.smallfiles.avgsize=16000000;  -- 16MB


-- 设置mapper处理的数据量
SET mapreduce.input.fileinputformat.split.maxsize=256000000;  -- 256MB
SET mapreduce.input.fileinputformat.split.minsize=128000000;  -- 128MB

--设置reducer数量
SET mapreduce.job.reduces=10;  -- 根据具体任务调整

--before(原始查询)
SELECT category_id, COUNT(*) as count
FROM sales_data
GROUPBY category_id;


--after(优化方案)

-- 方案1：开启Map端预聚合
SET hive.map.aggr=true;
SET hive.groupby.mapaggr.checkinterval=100000;

-- 方案2：开启倾斜数据优化(自动处理数据倾斜)
SET hive.groupby.skewindata=true;

-- 方案3：通过子查询打散数据(基本上等价于hive.groupby.skewindata的原理)
SELECT category_id, SUM(cnt) as count
FROM (
    SELECT category_id, COUNT(*) as cnt
    FROM sales_data
    GROUPBY category_id, CAST(RAND()*100ASINT)  -- 添加随机数打散
) t
GROUPBY category_id;

-- before(原始查询)
-- 方案0：原始ReduceJoin，也就是普通Join
SELECT a.user_id, a.order_id, b.user_name
FROM orders a
JOIN users b ON a.user_id = b.user_id;


-- after(优化方案)
-- 方案1：开启MapJoin 优化（适用于小表Join大表，小表的默认阈值是25M）
SET hive.auto.convert.join=true;
SET hive.mapjoin.smalltable.filesize=25000000;


-- 方案2：开启 SkewJoin 优化（自动处理数据倾斜，非倾斜key常规处理，倾斜key使MapJoin)
SET hive.optimize.skewjoin=true;
SET hive.skewjoin.key=100000;


-- 方案3： 手动处理，倾斜值隔离处理 (常用于null导致的倾斜)
WITH non_null_users AS (
    SELECT*
    FROM orders
    WHERE user_id ISNOT NULL
),
null_users AS (
    SELECT*
    FROM orders
    WHERE user_id ISNULL
)
SELECT*
FROM (
    -- 处理非NULL值
    SELECT a.*, b.user_name
    FROM non_null_users a
    JOIN users b ON a.user_id = b.user_id
    UNIONALL
    -- 处理NULL值
    SELECT a.*, NULLas user_name
    FROM null_users a
) t;



-- 方案4：手动处理，对随机数打散(加盐)
--步骤①: 对订单表中倾斜的user_id进行打散处理
WITH orders_expanded AS (
    SELECT
        -- 对倾斜的user_id增加随机后缀 0-9
        CASE
            WHEN user_id ='hot_user'THEN CONCAT(user_id, '_', CAST(RAND()*10ASINT))
            ELSE user_id 
        ENDAS new_user_id,
        user_id as original_user_id,  -- 保留原始user_id
        order_id
    FROM orders
),

--步骤②: 将用户表复制10份，每份对应一个随机后缀
users_expanded AS (
    -- 处理热点用户数据
    SELECT
        CONCAT(user_id, '_', suffix) AS new_user_id,
        user_id as original_user_id,
        user_name
    FROM users 
    CROSSJOIN (
        SELECTCAST(id AS STRING) AS suffix 
        FROM (
            SELECT0as id UNIONALLSELECT1UNIONALLSELECT2
            UNIONALLSELECT3UNIONALLSELECT4UNIONALLSELECT5
            UNIONALLSELECT6UNIONALLSELECT7UNIONALLSELECT8
            UNIONALLSELECT9
        ) numbers
    ) suffixes
    WHERE user_id ='hot_user'
    
    UNIONALL
    
    -- 处理非热点用户数据
    SELECT
        user_id AS new_user_id,
        user_id as original_user_id,
        user_name
    FROM users
    WHERE user_id !='hot_user'
)

--步骤③: 最终JOIN查询
SELECT
    o.original_user_id AS user_id,  -- 使用原始user_id
    o.order_id,
    u.user_name
FROM orders_expanded o
JOIN users_expanded u
ON o.new_user_id = u.new_user_id;

-- 创建桶表
CREATE TABLE bucket_user_logs (
    user_id INT,
    log_time STRING,
    action STRING
)
CLUSTERED BY (user_id) INTO4 BUCKETS
STORED AS ORC;

CREATE TABLE bucket_user_info (
    user_id INT,
    user_name STRING,
    user_age INT
)
CLUSTERED BY (user_id) INTO4 BUCKETS
STORED AS ORC;


-- 启用动态分区和分桶表插入
SET hive.exec.dynamic.partition=true;
SET hive.exec.dynamic.partition.mode=nonstrict;
SET hive.enforce.bucketing=true;

-- 插入数据到分区分桶表
INSERT INTOTABLE bucket_user_logs
SELECT user_id, log_time, action
FROM user_logs;

INSERT INTOTABLE bucket_user_info
SELECT user_id, user_name, user_age
FROM user_info;

-- 启用 BucketMapJoin 优化
SET hive.optimize.bucketmapjoin=true;
SET hive.optimize.bucketmapjoin.sortedmerge=true;

-- 基于桶的连接查询
SELECT l.user_id, l.log_time, l.action, i.user_name, i.user_age
FROM bucket_user_logs l
JOIN bucket_user_info i
ON l.user_id = i.user_id;

-- 原始方案： 大表直接 ordered by 取 topK 
SELECT sale_id, amount, product_id, sale_date
from  sales
ORDERBY amount DESC        -- 按销售额排序
LIMIT 100;                  -- 获取前100条记录


-- 优化方案： distribute by + sort by 后再 order by .
SELECT*
FROM (
    SELECT sale_id, amount, product_id, sale_date
    FROM sales
    DISTRIBUTE BY RAND()  -- 随机分布数据
    SORT BY amount DESC   -- 按销售额降序排序
    LIMIT 100           -- 先限制前1000条记录
) t
ORDERBY amount DESC        -- 最终按销售额降序排序
LIMIT 100;                  -- 获取前100条记录