import com.atlxl.ETLUtil;

/**
 * @author LXL
 * @create 2019-06-05 23:07
 */
public class Test {
    public static void main(String[] args) {
        System.out.println(ETLUtil.etlStr("SDNkMu8ZT68\tw00dy911\t630\tPeople & Blogs\t186\t10181\t3.49\t494\t257\trjnbgpPJUks\tkdjhkdfhsgksd"));
    }
}

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.atlxl</groupId>
    <artifactId>guliETL</artifactId>
    <version>1.0-SNAPSHOT</version>
    <dependencies>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-core</artifactId>
            <version>2.8.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-common</artifactId>
            <version>2.7.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>2.7.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-hdfs</artifactId>
            <version>2.7.2</version>
        </dependency>
    </dependencies>


</project>

package com.atlxl;

/**
 * @author LXL
 * @create 2019-06-05 22:27
 *
 * 1.过滤脏数据
 * 2.将类别字段中的空格
 * 3.替换关联视频中的分隔符
 */
public class ETLUtil {


    public static String etlStr(String line){

        //切割数据
        String[] split = line.split("\t");

        //1. 过滤脏数据
        if (split.length<9) return null;

        //2. 去掉类别字段中的空格
        split[3] = split[3].replaceAll(" ", "");

        //3.替换关联视频的分隔符
        StringBuffer sb = new StringBuffer();

        for (int i=0; i < split.length; i++) {
            if (i < 9) {
                if (i==split.length - 1){
                    sb.append(split[i]);
                }else {
                    sb.append(split[i]).append("\t");
                }
            }else {
                if (i==split.length - 1){
                    sb.append(split[i]);
                }else {
                    sb.append(split[i]).append("&");
                }
            }


        }

        return sb.toString();
    }

}

package com.atlxl;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/**
 * @author LXL
 * @create 2019-06-05 21:26
 */
public class ETlMapper extends Mapper<LongWritable, Text, Text, NullWritable> {


    private Text k = new Text();


    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {


        //1.获取一行数据
        String line = value.toString();


        //2.清洗数据
        String etlStr = ETLUtil.etlStr(line);


        //3.写出数据
        if (StringUtils.isBlank(etlStr)) {

            return;
        }
        k.set(etlStr);

        context.write(k, NullWritable.get());


    }
}

package com.atlxl;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

/**
 * @author LXL
 * @create 2019-06-05 23:22
 */
public class ETLDriver implements Tool{

    private Configuration configuration;


    public int run(String[] args) throws Exception {
        //1.获取job对象
        Job job = Job.getInstance(getConf());

        //2.封装driver类
        job.setJarByClass(ETLDriver.class);

        //3.关联Mapper类
        job.setMapperClass(ETlMapper.class);

        //4.Mapper输出的KV类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(NullWritable.class);

        //5.最终输出类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);

        //6.输入输出路径
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        job.setNumReduceTasks(0);

        //7.提交任务
        boolean result = job.waitForCompletion(true);
        return result ? 0 : 1;
    }

    public void setConf(Configuration conf) {
        configuration = conf;

    }

    public Configuration getConf() {
        return configuration;
    }

    public static void main(String[] args) throws Exception {

        int run = ToolRunner.run(new ETLDriver(), args);
        System.out.println(run);

    }

}

create table gulivideo_ori(
    videoId string,
    uploader string,
    age int,
    category array<string>,
    length int,
    views int,
    rate float,
    ratings int,
    comments int,
    relatedId array<string>)
row format delimited
fields terminated by "\t"
collection items terminated by "&"
stored as textfile;

create table gulivideo_user_ori(
    uploader string,
    videos int,
    friends int)
row format delimited
fields terminated by "\t"
stored as textfile;

create table gulivideo_orc(
    videoId string,
    uploader string,
    age int,
    category array<string>,
    length int,
    views int,
    rate float,
    ratings int,
    comments int,
    relatedId array<string>)
clustered by (uploader) into 8 buckets
row format delimited fields terminated by "\t"
collection items terminated by "&"
stored as orc;

create table gulivideo_user_orc(
    uploader string,
    videos int,
    friends int)
row format delimited
fields terminated by "\t"
stored as orc;

select * from gulivideo_orc order by views desc limit 10;

a.炸开类别
select videoId, category_name
from gulivideo_orc lateral view explode(category) category_t as category_name; t1

b.统计每种类别下的视频数
select category_name, count(*) hot
from t1 
group by category_name; t2

c.类别热度Top10
select category_name, hot 
from t2
order by hot desc
limit 10;

SELECT category_name, hot
FROM (
    SELECT category_name, COUNT(*) AS hot
    FROM (
        SELECT videoId, category_name
        FROM gulivideo_orc
            LATERAL VIEW explode(category) category_t AS category_name
    ) t1
    GROUP BY category_name
) t2
ORDER BY hot DESC
LIMIT 10;

select
    category_name as category,
    count(t1.videoId) as hot
from (
    select
        videoId,
        category_name
    from
        gulivideo_orc lateral view explode(category) t_catetory as category_name) t1
group by
    t1.category_name
order by
    hot
desc limit
    10;

SELECT category_name, COUNT(*) AS hot
FROM (
    SELECT videoId, category_name
    FROM gulivideo_orc
        LATERAL VIEW explode(category) category_t AS category_name
) t1
GROUP BY category_name
ORDER BY hot DESC
LIMIT 10;

a.统计观看数Top20所属类别
select videoId,views,category
from gulivideo_orc
order by views desc
limit 20; t1 

b.前20视频的类别
select videoId, category_name
from t1
lateral view explode(category) category_t as category_name; t2

c.去重
select videoId, distinct(t2.category_name)
from t2

select
    category_name as category,
    count(t2.videoId) as hot_with_views
from (
    select
        videoId,
        category_name
    from (
        select
            *
        from
            gulivideo_orc
        order by
            views
        desc limit
            20) t1 lateral view explode(category) t_catetory as category_name) t2
group by
    category_name
order by
    hot_with_views
desc;

a.统计观看数前50的视频&关联视频
select
    videoId, views, category, relatedId
from
    gulivideo_orc
order by
    views DESC
limit 50; t1


b.炸开关联视频id 去重
select DISTINCT(relatedId_name)
from t1
lateral view explode(relatedId) relatedId_t as relatedId_name; t2



c.关联视频所属类别
select *
from t2 join 
gulivideo_orc t3
on t2.relatedId_name=t3.videoId; t4


d.炸开关联视频的类别
select * from t4
lateral view explode(category) category_t as category_name; t5


e.统计类别个数
SELECT category_name,count(*) hot
from t5
GROUP BY category_name; t6

f.统计类别Rank
SELECT * FROM t6 order by hot desc;

SELECT *
FROM (
    SELECT category_name, COUNT(*) AS hot
    FROM (
        SELECT *
        FROM (
            SELECT *
            FROM (
                SELECT DISTINCT relatedId_name
                FROM (
                    SELECT videoId, views, category, relatedId
                    FROM gulivideo_orc
                    ORDER BY views DESC
                    LIMIT 50
                ) t1
                    LATERAL VIEW explode(relatedId) relatedId_t AS relatedId_name
            ) t2
                JOIN gulivideo_orc t3 ON t2.relatedId_name = t3.videoId
        ) t4
            LATERAL VIEW explode(category) category_t AS category_name
    ) t5
    GROUP BY category_name
) t6
ORDER BY hot DESC;

create table gulivideo_category(
    videoId string,
    uploader string,
    age int,
    categoryId string,
    length int,
    views int,
    rate float,
    ratings int,
    comments int,
    relatedId array<string>)
row format delimited
fields terminated by "\t"
collection items terminated by "&"
stored as orc;

insert into table gulivideo_category  
    select
        videoId,
        uploader,
        age,
        categoryId,
        length,
        views,
        rate,
        ratings,
        comments,
        relatedId
    from
        gulivideo_orc lateral view explode(category) catetory as categoryId;

select
    videoId,
    views
from
    gulivideo_category
where
    categoryId = "Music"
order by
    views
desc limit
    10;

select
    videoId,
    views,
    ratings
from
    gulivideo_category
where
    categoryId = "Music"
order by
    ratings
desc limit
    10;

a.上传视频数钱20的人
SELECT uploader FROM gulivideo_user_orc ORDER BY videos DESC LIMIT 10; t1

b.这10个人所上传的视频
SELECT * FROM t1
jion gulivideo_orc t2
on t1.uploader = t2.uploader; t3

c.观看次数在前20的视频
SELECT *
FROM t3
ORDER BY views
DESC
LIMIT 20;


SELECT *
FROM (SELECT * FROM (SELECT uploader,videos FROM gulivideo_user_orc ORDER BY videos DESC LIMIT 10) t1
join gulivideo_orc t2
on t1.uploader = t2.uploader) t3
ORDER BY views
DESC
LIMIT 20;


SELECT *
FROM (
    SELECT views, videoId, t1.uploader
    FROM (
        SELECT uploader, videos
        FROM gulivideo_user_orc
        ORDER BY videos DESC
        LIMIT 10
    ) t1
        JOIN gulivideo_orc t2 ON t1.uploader = t2.uploader
) t3
ORDER BY views DESC
LIMIT 20;

select
    t1.*
from (
    select
        videoId,
        categoryId,
        views,
        row_number() over(partition by categoryId order by views desc) rank from gulivideo_category) t1
where
    rank <= 10;

9.统计出视频观看数最高的20个视频的所属类别以及类别包含Top20视频的个数

select
    category_name as category,
    count(t2.videoId) as hot_with_views
from (
    select
        videoId,
        category_name
    from (
        select
            *
        from
            gulivideo_orc
        order by
            views
        desc limit
            20) t1 lateral view explode(category) t_catetory as category_name) t2
group by
    category_name
order by
    hot_with_views
desc;