Hive拓展项目之Youtube

300

create table company_info(
    personId string,
    company string,
    money float
)row format delimited fields terminated by "\t"
load data local inpath “company_info.txt” into table company_info;

select * from company_info order by money desc;

select * from company_info distribute by personId sort by personId, money desc;

select * from company_info distribute by personId sort by personId;
等价于
select * from compnay_info cluster by personId;

A

create table person_info(
name string, 
constellation string, 
blood_type string) 
row format delimited fields terminated by "\t";
load data local inpath “person_info.tsv” into table person_info;

select
    t1.base,
    concat_ws('|', collect_set(t1.name)) name
from
    (select
        name,
        concat(constellation, ",", blood_type) base
    from
        person_info) t1
group by
    t1.base;

战争,动作,灾难

create table movie_info(
    movie string, 
    category array<string>) 
row format delimited fields terminated by "\t"
collection items terminated by ",";
load data local inpath "movie_info.tsv" into table movie_info;

select
    movie,
    category_name 
from 
    movie_info lateral view explode(category) table_tmp as category_name;

create table music(
    id int,
    name string,
    size float)
row format delimited 
fields terminated by "\t"
clustered by (id) into 4 buckets;

create table music2(
    id int,
    name string,
    size float)
partitioned by (date string)
clustered by (id) sorted by(size) into 4 bucket
row format delimited 
fields terminated by "\t";
 
load data local inpath 'demo/music.txt' into table music2 partition(date='2017-08-30');

相关视频的id，最多20个

int

<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.z</groupId>
       <artifactId>youtube</artifactId>
       <version>0.0.1-SNAPSHOT</version>
       <packaging>jar</packaging>
 
       <name>youtube</name>
       <url>http://maven.apache.org</url>
 
       <properties>
              <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
       </properties>
 
       <repositories>
              <repository>
                     <id>centor</id>
                     <url>http://central.maven.org/maven2/</url>
              </repository>
       </repositories>
 
       <dependencies>
              <dependency>
                     <groupId>junit</groupId>
                     <artifactId>junit</artifactId>
                     <version>3.8.1</version>
                     <scope>test</scope>
              </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-yarn-server-resourcemanager</artifactId>
                     <version>2.7.2</version>
              </dependency>
       </dependencies>
</project>

package com.z.youtube.util;
 
public class ETLUtils {
       /**
        * 1、过滤不合法数据
        * 2、去掉&符号左右两边的空格
        * 3、\t换成&符号
        * @param ori
        * @return
        */
       public static String getETLString(String ori){
              String[] splits = ori.split("\t");
              //1、过滤不合法数据
              if(splits.length < 9) return null;
              //2、去掉&符号左右两边的空格
              splits[3] = splits[3].replaceAll(" ", "");
             
              StringBuilder sb = new StringBuilder();
              //3、\t换成&符号
              for(int i = 0; i < splits.length; i++){
                     sb.append(splits[i]);
                     if(i < 9){
                            if(i != splits.length - 1){
                                   sb.append("\t");
                            }
                     }else{
                            if(i != splits.length - 1){
                                   sb.append("&");
                            }
                     }
              }
              return sb.toString();
       }
}

package com.z.youtube.mr.etl;
 
import java.io.IOException;
 
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
 
import com.z.youtube.util.ETLUtil;
 
public class VideoETLMapper extends Mapper<Object, Text, NullWritable, Text>{
       Text text = new Text();
      
       @Override
       protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
              String etlString = ETLUtil.oriString2ETLString(value.toString());
             
              if(StringUtils.isBlank(etlString)) return;
             
              text.set(etlString);
              context.write(NullWritable.get(), text);
       }
      
 
}

package com.z.youtube.mr.etl;
 
import java.io.IOException;
 
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
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;
 
public class VideoETLRunner implements Tool {
       private Configuration conf = null;
 
       @Override
       public void setConf(Configuration conf) {
              this.conf = conf;
       }
 
       @Override
       public Configuration getConf() {
 
              return this.conf;
       }
 
       @Override
       public int run(String[] args) throws Exception {
              conf = this.getConf();
              conf.set("inpath", args[0]);
              conf.set("outpath", args[1]);
 
              Job job = Job.getInstance(conf, "youtube-video-etl");
             
              job.setJarByClass(VideoETLRunner.class);
             
              job.setMapperClass(VideoETLMapper.class);
              job.setMapOutputKeyClass(NullWritable.class);
              job.setMapOutputValueClass(Text.class);
              job.setNumReduceTasks(0);
             
              this.initJobInputPath(job);
              this.initJobOutputPath(job);
             
              return job.waitForCompletion(true) ? 0 : 1;
       }
 
       private void initJobOutputPath(Job job) throws IOException {
              Configuration conf = job.getConfiguration();
              String outPathString = conf.get("outpath");
             
              FileSystem fs = FileSystem.get(conf);
             
              Path outPath = new Path(outPathString);
              if(fs.exists(outPath)){
                     fs.delete(outPath, true);
              }
             
              FileOutputFormat.setOutputPath(job, outPath);
             
       }
 
       private void initJobInputPath(Job job) throws IOException {
              Configuration conf = job.getConfiguration();
              String inPathString = conf.get("inpath");
             
              FileSystem fs = FileSystem.get(conf);
             
              Path inPath = new Path(inPathString);
              if(fs.exists(inPath)){
                     FileInputFormat.addInputPath(job, inPath);
              }else{
                     throw new RuntimeException("HDFS中该文件目录不存在：" + inPathString);
              }
       }
 
       public static void main(String[] args) {
              try {
                     int resultCode = ToolRunner.run(new VideoETLRunner(), args);
                     if(resultCode == 0){
                            System.out.println("Success!");
                     }else{
                            System.out.println("Fail!");
                     }
                     System.exit(resultCode);
              } catch (Exception e) {
                     e.printStackTrace();
                     System.exit(1);
              }
       }
}

$ bin/yarn jar ~/softwares/jars/youtube-0.0.1-SNAPSHOT.jar \
com.z.youtube.etl.ETLYoutubeVideosRunner \
/youtube/video/2008/0222 \
/youtube/output/video/2008/0222

create table youtube_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 youtube_user_ori(
    uploader string,
    videos int,
    friends int)
clustered by (uploader) into 24 buckets
row format delimited 
fields terminated by "\t" 
stored as textfile;

create table youtube_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 youtube_user_orc(
    uploader string,
    videos int,
    friends int)
clustered by (uploader) into 24 buckets 
row format delimited 
fields terminated by "\t" 
stored as orc;

load data inpath "/youtube/output/video/2008/0222" into table youtube_ori;

load data inpath "/youtube/user/2008/0903" into table youtube_user_ori;

insert into table youtube_orc select * from youtube_ori;

insert into table youtube_user_orc select * from youtube_user_ori;

select
    videoId, 
    uploader, 
    age, 
    category, 
    length, 
    views, 
    rate, 
    ratings, 
    comments 
from 
    youtube_orc 
order by 
    views 
desc limit 
    10;

select
    category_name as category, 
    count(t1.videoId) as hot 
from (
    select 
        videoId,
        category_name 
    from 
        youtube_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 as category, 
    count(t2.videoId) as hot_with_views 
from (
    select 
        videoId, 
        category_name 
    from (
        select 
            * 
        from 
            youtube_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;

t1:观看数前50的视频
select 
    * 
from 
    youtube_orc 
order by 
    views 
desc limit 
    50;

t2:将相关视频的id进行列转行操作
select 
    explode(relatedId) as videoId 
from 
       t1;

t5:得到两列数据，一列是category，一列是之前查询出来的相关视频id
(select 
    distinct(t2.videoId), 
    t3.category 
from 
    t2
inner join 
    youtube_orc t3 on t2.videoId = t3.videoId) t4 lateral view explode(category) t_catetory as category_name;

select
    category_name as category, 
    count(t5.videoId) as hot 
from (
    select 
        videoId, 
        category_name 
    from (
        select 
            distinct(t2.videoId), 
            t3.category 
        from (
            select 
                explode(relatedId) as videoId 
            from (
                select 
                    * 
                from 
                    youtube_orc 
                order by 
                    views 
                desc limit 
                    50) t1) t2 
        inner join 
            youtube_orc t3 on t2.videoId = t3.videoId) t4 lateral view explode(category) t_catetory as category_name) t5
group by 
    category_name 
order by 
    hot 
desc;

create table youtube_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 youtube_category 
    select 
        videoId,
        uploader,
        age,
        categoryId,
        length,
        views,
        rate,
        ratings,
        comments,
        relatedId 
    from 
        youtube_orc lateral view explode(category) catetory as categoryId;

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

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

select
    * 
from 
    youtube_user_orc 
order by 
    videos 
desc limit 
    10;

select
    t2.videoId, 
    t2.views,
    t2.ratings,
    t1.videos,
    t1.friends 
from (
    select 
        * 
    from 
        youtube_user_orc 
    order by 
        videos desc 
    limit 
        10) t1 
join 
    youtube_orc t2
on 
    t1.uploader = t2.uploader 
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 youtube_category) t1 
where 
    rank <= 10;

<property>
       <name>yarn.scheduler.maximum-allocation-mb</name>
       <value>2048</value>
</property>
<property>
     <name>yarn.scheduler.minimum-allocation-mb</name>
     <value>2048</value>
</property>
<property>
       <name>yarn.nodemanager.vmem-pmem-ratio</name>
       <value>2.1</value>
</property>
<property>
       <name>mapred.child.java.opts</name>
       <value>-Xmx1024m</value>
</property>