经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » 大数据/云/AI » Hadoop » 查看文章
ETL项目2:大数据清洗,处理:使用MapReduce进行离线数据分析并报表显示完整项目
来源：cnblogs　　作者：symkmk123　　时间：2019/1/2 9:38:02　　对本文有异议

ETL项目2:大数据清洗,处理:使用MapReduce进行离线数据分析并报表显示完整项目

思路同我之前的博客的思路 https://www.cnblogs.com/symkmk123/p/10197467.html

但是数据是从web访问的数据

avro第一次过滤

观察数据的格式,我们主要分析第四个字段的数据.发现有.css , .jpg .png等等等无效的数据.

通过观察数据发现有效数据都不带 . , 所以第一次过滤写入avro总表里的数据一次过滤后的有效数据,不包含 .css , .jpg , .png 这样的数据

同时count持久化到mysql

orc1:海牛的topics 最受欢迎的top10

通过观察发现这个需求的有效url是 /topics/数字的 所以在第一次过滤的数据的基础上的正则就是

这种保留下来的也只是/topics/数字这种格式,方便用 hql统计结果

 上代码

  1. //Text2Avro
  2. package mrrun.hainiuetl;
  4. import java.io.IOException;
  5. import java.text.ParseException;
  6. import java.text.SimpleDateFormat;
  7. import java.util.Date;
  8. import java.util.Locale;
  10. import org.apache.avro.Schema;
  11. import org.apache.avro.generic.GenericData;
  12. import org.apache.avro.generic.GenericRecord;
  13. import org.apache.avro.mapred.AvroKey;
  14. import org.apache.avro.mapreduce.AvroJob;
  15. import org.apache.avro.mapreduce.AvroKeyOutputFormat;
  16. import org.apache.hadoop.conf.Configuration;
  17. import org.apache.hadoop.io.LongWritable;
  18. import org.apache.hadoop.io.NullWritable;
  19. import org.apache.hadoop.io.Text;
  20. import org.apache.hadoop.mapreduce.Counter;
  21. import org.apache.hadoop.mapreduce.CounterGroup;
  22. import org.apache.hadoop.mapreduce.Counters;
  23. import org.apache.hadoop.mapreduce.Job;
  24. import org.apache.hadoop.mapreduce.Mapper;
  25. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  26. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  28. import mrrun.base.BaseMR;
  30. public class Text2Avro extends BaseMR
  31. {
  32.     public static Schema schema = null;
  33.     
  34.     public static Schema.Parser parse = new Schema.Parser();
  35.     
  36. public static class Text2AvroMapper extends Mapper<LongWritable, Text, AvroKey<GenericRecord>, NullWritable>
  37. {
  38.         
  39.     
  40.     
  41.         @Override
  42.         protected void setup(Mapper<LongWritable, Text, AvroKey<GenericRecord>, NullWritable>.Context context)
  43.                 throws IOException, InterruptedException {
  44.             //根据user_install_status.avro文件内的格式，生成指定格式的schema对象
  45.             schema = parse.parse(Text2Avro.class.getResourceAsStream("/hainiu.avro"));
  46.             
  47.         }
  48.         @Override
  49.         protected void map(LongWritable key, Text value,Context context)
  50.                 throws IOException, InterruptedException {
  51.             String line = value.toString();
  52.             
  53.             String[] splits = line.split("\001");
  54.             if(splits == null || splits.length != 10){
  55.                 System.out.println("==============");
  56.                 System.out.println(value.toString());
  57.                 context.getCounter("etl_err", "bad line num").increment(1L);
  58.                 return;
  59.             }
  60.             
  61. //            System.out.println(util.getIpArea("202.8.77.12"));
  62.             String uip1 = splits[0];
  63.             String uip =IPUtil.getip(uip1);
  64.             
  65.             String datetime = splits[2];
  66.             StringBuilder sb=new StringBuilder(datetime);
  67.             
  68.             SimpleDateFormat sdf=new SimpleDateFormat("dd/MMM/yyyy:HH:mm:ss",Locale.ENGLISH);
  69.             String sy=sb.toString();
  70.             Date myDate = null;
  71.             try
  72.             {
  73.                 myDate = sdf.parse(sy);
  74.             } catch (ParseException e)
  75.             {
  76.                 // TODO Auto-generated catch block
  77.                 e.printStackTrace();
  78.             }
  79.             
  80.             SimpleDateFormat sdf2=new SimpleDateFormat("yyyyMMddHHmmss");
  81.             //System.out.println(myDate);
  82.             String format = sdf2.format(myDate);
  83.             //GET /categories/8?filter=recent&page=12 HTTP/1.1
  84.             String url1 = splits[3];
  85.             StringBuilder sb2=new StringBuilder(url1);
  86.             
  87.             String url = sb2.toString();
  88.             String method="";
  89.             String top="";
  90.             String top1="";
  91.             String http="";
  92.             if(url!=null)
  93.             {
  94.                 String[] s = url.split(" ");
  95.                 if(s.length==3)
  96.                 {
  97.                     method=s[0];
  98.                     http=s[2];
  99.                     
  100.                     top1=s[1];
  101.                     if(top1.contains("."))
  102.                     {
  103.                         context.getCounter("etl_err", "no line num").increment(1L);
  104.                         return;
  105.                     }
  106.                     else
  107.                     {
  108.                         top=top1;
  109.                     }
  110.                 }
  111.             }
  112.             
  113.             String status1 = splits[4];
  114.             String status2 = splits[5];
  115.             String post = splits[6];
  116.             String from = splits[7];
  117.             String usagent1 = splits[8];
  118.             StringBuilder sb3=new StringBuilder(usagent1);
  119.         
  120.             String usagent = sb3.toString();
  121.             
  122.             
  123.             //根据创建的Schema对象，创建一行的对象
  124.             GenericRecord record = new GenericData.Record(Text2Avro.schema);
  125.             record.put("uip", uip);
  126.             record.put("datetime", format);
  127.             record.put("method", method);
  128.             record.put("http", http);
  129.             record.put("top", top);
  130.             record.put("from", from);
  131.             record.put("status1", status1);
  132.             record.put("status2", status2);
  133.             record.put("post", post);
  134.             record.put("usagent", usagent);
  135.             
  136.             context.getCounter("etl_good", "good line num").increment(1L);
  137.             System.out.println(uip+"    "+format+"    "+top+"    "+from+"    "+post+"    "+usagent+"    "+status1+"    "+status2+"    "+http);
  138.             
  139.             
  140.             context.write(new AvroKey<GenericRecord>(record), NullWritable.get());
  141.             
  142.         
  143.         }
  144.     }
  145.     
  146.     
  147.     
  148.     
  149.     @Override
  150.     public Job getJob(Configuration conf) throws IOException {
  151. //        // 开启reduce输出压缩
  152. //        conf.set(FileOutputFormat.COMPRESS, "true");
  153. //        // 设置reduce输出压缩格式
  154. //        conf.set(FileOutputFormat.COMPRESS_CODEC, SnappyCodec.class.getName());
  155.                 
  156.         Job job = Job.getInstance(conf, getJobNameWithTaskId());
  158.         job.setJarByClass(Text2Avro.class);
  159.         
  160.         job.setMapperClass(Text2AvroMapper.class);
  161.         
  162.         job.setMapOutputKeyClass(AvroKey.class);
  163.         job.setMapOutputValueClass(NullWritable.class);
  164.         
  165. //        无reduce
  166.         job.setNumReduceTasks(0);
  167.         
  168.         //设置输出的format
  169.         job.setOutputFormatClass(AvroKeyOutputFormat.class);
  170.         
  171.         //根据user_install_status.avro文件内的格式，生成指定格式的schema对象
  172.         schema = parse.parse(Text2Avro.class.getResourceAsStream("/hainiu.avro"));
  173.         
  174.         //设置avro文件的输出
  175.         AvroJob.setOutputKeySchema(job, schema);
  176.         
  177.         FileInputFormat.addInputPath(job, getFirstJobInputPath());
  178.         
  179.         FileOutputFormat.setOutputPath(job, getJobOutputPath(getJobNameWithTaskId()));
  180.         
  181.         
  182.          
  183.         return job;
  185.     }
  187.     @Override
  188.     public String getJobName() {
  191.         return "etltext2avro";
  193.     }
  195. }
  1. //Avro2Orc_topic10
  2. package mrrun.hainiuetl;
  4. import java.io.IOException;
  5. import java.util.regex.Matcher;
  6. import java.util.regex.Pattern;
  8. import org.apache.avro.Schema;
  9. import org.apache.avro.generic.GenericRecord;
  10. import org.apache.avro.mapred.AvroKey;
  11. import org.apache.avro.mapreduce.AvroJob;
  12. import org.apache.avro.mapreduce.AvroKeyInputFormat;
  13. import org.apache.hadoop.conf.Configuration;
  14. import org.apache.hadoop.hive.ql.io.orc.CompressionKind;
  15. import org.apache.hadoop.hive.ql.io.orc.OrcNewOutputFormat;
  16. import org.apache.hadoop.io.NullWritable;
  17. import org.apache.hadoop.io.Writable;
  18. import org.apache.hadoop.mapreduce.Job;
  19. import org.apache.hadoop.mapreduce.Mapper;
  20. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  21. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  23. import mrrun.base.BaseMR;
  24. import mrrun.util.OrcFormat;
  25. import mrrun.util.OrcUtil;
  27. public class Avro2Orc_topic10 extends BaseMR {
  28.     public static Schema schema = null;
  29.     
  30.     public static Schema.Parser parse = new Schema.Parser();
  31.     
  33.     public static class Avro2OrcMapper extends Mapper<AvroKey<GenericRecord>, NullWritable, NullWritable, Writable>{
  34.         OrcUtil orcUtil = new OrcUtil();
  35.         
  36.         @Override
  37.         protected void setup(Context context)
  38.                 throws IOException, InterruptedException {
  39.             orcUtil.setWriteOrcInspector(OrcFormat.etlorcSchema_topic10);
  40.             
  41.         }
  42.         
  43.         @Override
  44.         protected void map(AvroKey<GenericRecord> key, NullWritable value,Context context)
  45.                 throws IOException, InterruptedException {
  46.             //得到一行的对象
  47.             GenericRecord datum = key.datum();
  48.             
  49.             String uip = (String) datum.get("uip");
  50.             String datetime = (String) datum.get("datetime");
  51.             //String method = (String) datum.get("method");
  52.             //String http = (String) datum.get("http");
  53.             String top1 = (String) datum.get("top");
  54.             String top="";
  55.             String regex="/topics/\\d+";
  56.             Pattern pattern=Pattern.compile(regex);
  57.             Matcher matcher=pattern.matcher(top1);
  58.             if(matcher.find())
  59.             {
  60.                  top=matcher.group();
  61.             }
  62.             else
  63.             {
  64.                 context.getCounter("etl_err", "notopics line num").increment(1L);
  65.                 return;
  66.             }
  67.             
  68.             
  69.             
  70.             //orcUtil.addAttr(uip,datetime,method,http,uid,country,status1,status2,usagent);
  71.             orcUtil.addAttr(uip,datetime,top);
  72.             
  73.             Writable w = orcUtil.serialize();
  74.             context.getCounter("etl_good", "good line num").increment(1L);
  75.             System.out.println(uip+"    "+top);
  76.             
  77.             context.write(NullWritable.get(), w);
  78.             
  79.         }
  80.         
  81.     }
  82.     
  84.     
  85.     @Override
  86.     public Job getJob(Configuration conf) throws IOException {
  87.         
  88.         //关闭map的推测执行，使得一个map处理 一个region的数据
  89.         conf.set("mapreduce.map.spedulative", "false");
  90.         //设置orc文件snappy压缩
  91.         conf.set("orc.compress", CompressionKind.SNAPPY.name());
  92.         //设置orc文件 有索引
  93.         conf.set("orc.create.index", "true");
  94.                 
  95.         Job job = Job.getInstance(conf, getJobNameWithTaskId());
  97.         job.setJarByClass(Avro2Orc_topic10.class);
  98.         
  99.         job.setMapperClass(Avro2OrcMapper.class);
  100.         
  101.         job.setMapOutputKeyClass(NullWritable.class);
  102.         job.setMapOutputValueClass(Writable.class);
  104.         
  105. //        无reduce
  106.         job.setNumReduceTasks(0);
  107.         
  108.         job.setInputFormatClass(AvroKeyInputFormat.class);
  109.         
  110.         //根据user_install_status.avro文件内的格式，生成指定格式的schema对象
  111.         schema = parse.parse(Avro2Orc_topic10.class.getResourceAsStream("/hainiu.avro"));
  112.         
  113.         AvroJob.setInputKeySchema(job, schema);
  114.         
  115.     
  116.         job.setOutputFormatClass(OrcNewOutputFormat.class);
  117.         
  118.         
  119.         FileInputFormat.addInputPath(job, getFirstJobInputPath());
  120.         
  121.         FileOutputFormat.setOutputPath(job, getJobOutputPath(getJobNameWithTaskId()));
  122.         return job;
  124.     }
  126.     @Override
  127.     public String getJobName() {
  130.         return "etlAvro2Orc_topic10";
  132.     }
  134. }
  1. //Text2AvroJob 
  2. package mrrun.hainiuetl;
  5. import java.text.SimpleDateFormat;
  6. import java.util.Date;
  8. import org.apache.hadoop.conf.Configuration;
  9. import org.apache.hadoop.conf.Configured;
  10. import org.apache.hadoop.mapreduce.Counter;
  11. import org.apache.hadoop.mapreduce.CounterGroup;
  12. import org.apache.hadoop.mapreduce.Counters;
  13. import org.apache.hadoop.mapreduce.Job;
  14. import org.apache.hadoop.mapreduce.lib.jobcontrol.ControlledJob;
  15. import org.apache.hadoop.mapreduce.lib.jobcontrol.JobControl;
  16. import org.apache.hadoop.util.Tool;
  17. import org.apache.hadoop.util.ToolRunner;
  19. import mrrun.util.JobRunResult;
  20. import mrrun.util.JobRunUtil;
  22. public class Text2AvroJob extends Configured implements Tool{
  23.     
  24.     @Override
  25.     public int run(String[] args) throws Exception {
  26.         //获取Configuration对象
  27.         Configuration conf = getConf();
  28.         
  29.         
  30.         //创建任务链对象
  31.         JobControl jobc = new JobControl("etltext2avro");
  32.         
  33.         Text2Avro avro = new Text2Avro();
  34.         
  35.         //只需要赋值一次就行
  36.         avro.setConf(conf);
  37.         
  38.         ControlledJob orcCJob = avro.getControlledJob();
  39.         
  40.         Job job = orcCJob.getJob();
  41.         job.waitForCompletion(true);
  42.         
  43.         JobRunResult result = JobRunUtil.run(jobc);
  44.         result.setCounters("etl1", orcCJob.getJob().getCounters());
  45.         
  46.         result.print(true);
  47.         
  48.         Counters counterMap = result.getCounterMap("etl1");
  49.          CounterGroup group1 = counterMap.getGroup("etl_good");
  50.          CounterGroup group2 = counterMap.getGroup("etl_err");
  51.         
  52.          Counter good = group1.findCounter("good line num");
  53.          Counter bad = group2.findCounter("bad line num");
  54.         System.out.println("\t\t"+good.getDisplayName()+"  =  "+good.getValue());
  55.         System.out.println("\t\t"+bad.getDisplayName()+"  =  "+bad.getValue());
  56.         System.out.println("=======+++++++++====");
  57.         
  59.         Date date=new Date();
  60.         SimpleDateFormat sdf3=new SimpleDateFormat("yyyyMMdd");
  61.         String format2 = sdf3.format(date);
  62.         Results results=new Results();
  63.         long bad_num = bad.getValue();
  64.         long good_num = good.getValue();
  65.     
  66.         long total_num=bad_num+good_num;
  67.         results.setBad_num(bad_num);
  68.         results.setGood_num(good_num);
  69.         
  70.         results.setTotal_num(total_num);
  71.         results.setDay(format2);
  72.         double d=bad_num*1.0/total_num*1.0;
  73.         
  74.         results.setBad_rate(d);
  75.         
  76.         
  77.         System.out.println((double)((double)bad_num/(double)total_num));
  78.         
  79.         DAO dao=new DAO();
  80.         if(dao.getday(format2)!=null)
  81.         {
  82.             Results getday = dao.getday(format2);
  83.             Long bad_num2 = getday.getBad_num();
  84.             Long good_num2 = getday.getGood_num();
  85.             Long total_num2 = getday.getTotal_num();
  86.             getday.setDay(format2);
  87.             getday.setBad_num(bad_num2+bad_num);
  88.             getday.setGood_num(good_num2+good_num);
  89.             
  90.             getday.setTotal_num(total_num2+total_num);
  91.             double badrate=(bad_num2+bad_num)*1.0/(total_num2+total_num)*1.0;
  92.         
  93.             
  94.             getday.setBad_rate(badrate);
  95.         
  96.             dao.update(getday);
  97.         }
  98.         else
  99.         {
  100.             dao.insert(results);
  101.         }
  102.         
  103.         jobc.addJob(orcCJob);
  104.         
  105.         return 0;
  106.         
  107.     }
  109.     
  110.     public static void main(String[] args) throws Exception {
  111. //        -Dtask.id=1226 -Dtask.input.dir=/tmp/avro/input_hainiuetl -Dtask.base.dir=/tmp/avro
  112.         System.exit(ToolRunner.run(new Text2AvroJob(), args));
  113.     }
  115. }

放一个

自动化脚本思路同第一个ETL项目

直接放代码

  1. yitiaolong.sh
  3. #！/bin/bash
  4. source /etc/profile
  5. mmdd=`date -d 1' days ago' +%m%d`
  6. yymm=`date -d 1' days ago' +%Y%m`
  7. dd=`date -d 1' days ago' +%d`
  8. /usr/local/hive/bin/hive -e "use suyuan09;alter table etlavrosy add IF NOT EXISTS partition(month='${yymm}',day='${dd}');"
  9. /usr/local/hive/bin/hive -e "use suyuan09;alter table hainiuetltopics10_orc add IF NOT EXISTS partition(month='${yymm}',day='${dd}');"
  10. /usr/local/hive/bin/hive -e "use suyuan09;alter table hainiuetlcategories10_orc add IF NOT EXISTS partition(month='${yymm}',day='${dd}');"
  11. /usr/local/hive/bin/hive -e "use suyuan09;alter table hainiuetlspider_orc add IF NOT EXISTS partition(month='${yymm}',day='${dd}');"
  12. /usr/local/hive/bin/hive -e "use suyuan09;alter table hainiuetlip_orc add IF NOT EXISTS partition(month='${yymm}',day='${dd}');"
  13. /usr/local/hive/bin/hive -e "use suyuan09;alter table hainiuetlindex5_orc add IF NOT EXISTS partition(month='${yymm}',day='${dd}');"
  16. #3-4运行mr
  17. hdfs_path1=/user/hainiu/data/hainiuetl/input/${yymm}/${dd}
  18. avro_path1=/user/suyuan09/hainiuetl/hainiuavro/${yymm}/${dd}
  19. `/usr/local/hadoop/bin/hadoop  jar /home/suyuan09/etl/hainiu/jar/181210_hbase-1.0.0-symkmk123.jar etltext2avro -Dtask.id=${mmdd} -Dtask.input.dir=${hdfs_path1} -Dtask.base.dir=${avro_path1}`
  21. #orctopics10mr.sh
  23. avro_path2=/user/suyuan09/hainiuetl/hainiuavro/${yymm}/${dd}/etltext2avro_${mmdd}/part-*.avro
  24. orc_path2=/user/suyuan09/hainiuetl/orctopics10/${yymm}/${dd}
  25. `/usr/local/hadoop/bin/hadoop  jar /home/suyuan09/etl/hainiu/jar/181210_hbase-1.0.0-symkmk123.jar etlavro2orc_topic10 -Dtask.id=${mmdd} -Dtask.input.dir=${avro_path2} -Dtask.base.dir=${orc_path2}`
  27. #orccategories10mr.sh
  30. avro_path3=/user/suyuan09/hainiuetl/hainiuavro/${yymm}/${dd}/etltext2avro_${mmdd}/part-*.avro
  31. orc_path3=/user/suyuan09/hainiuetl/orccategories10/${yymm}/${dd}
  32. `/usr/local/hadoop/bin/hadoop  jar /home/suyuan09/etl/hainiu/jar/181210_hbase-1.0.0-symkmk123.jar etlavro2orc_categories10 -Dtask.id=${mmdd} -Dtask.input.dir=${avro_path3} -Dtask.base.dir=${orc_path3}`
  34. #orcspidermr.sh
  36. avro_path4=/user/suyuan09/hainiuetl/hainiuavro/${yymm}/${dd}/etltext2avro_${mmdd}/part-*.avro
  37. orc_path4=/user/suyuan09/hainiuetl/orcspider/${yymm}/${dd}
  38. `/usr/local/hadoop/bin/hadoop  jar /home/suyuan09/etl/hainiu/jar/181210_hbase-1.0.0-symkmk123.jar etlavro2orc_spider -Dtask.id=${mmdd} -Dtask.input.dir=${avro_path4} -Dtask.base.dir=${orc_path4}`
  40. #orcipmr.sh
  42. avro_path5=/user/suyuan09/hainiuetl/hainiuavro/${yymm}/${dd}/etltext2avro_${mmdd}/part-*.avro
  43. orc_path5=/user/suyuan09/hainiuetl/orcip/${yymm}/${dd}
  44. `/usr/local/hadoop/bin/hadoop  jar /home/suyuan09/etl/hainiu/jar/181210_hbase-1.0.0-symkmk123.jar etlavro2orc_ip -Dtask.id=${mmdd} -Dtask.input.dir=${avro_path5} -Dtask.base.dir=${orc_path5}`
  46. #orcindex5mr.sh
  48. avro_path6=/user/suyuan09/hainiuetl/hainiuavro/${yymm}/${dd}/etltext2avro_${mmdd}/part-*.avro
  49. orc_path6=/user/suyuan09/hainiuetl/orcindex5/${yymm}/${dd}
  50. `/usr/local/hadoop/bin/hadoop  jar /home/suyuan09/etl/hainiu/jar/181210_hbase-1.0.0-symkmk123.jar etlavro2orc_index5 -Dtask.id=${mmdd} -Dtask.input.dir=${avro_path6} -Dtask.base.dir=${orc_path6}`
  52. #把orc挪到分区目录  
  53. #orc2etl.sh
  54.  
  55. /usr/local/hadoop/bin/hadoop fs -cp hdfs://ns1/user/suyuan09/hainiuetl/orctopics10/${yymm}/${dd}/etlAvro2Orc_topic10_${mmdd}/part-*  hdfs://ns1/user/suyuan09/etlorc/hainiuetltopics10_orc/month=${yymm}/day=${dd}
  56. /usr/local/hadoop/bin/hadoop fs -cp hdfs://ns1/user/suyuan09/hainiuetl/orccategories10/${yymm}/${dd}/etlAvro2Orc_categories10_${mmdd}/part-*  hdfs://ns1/user/suyuan09/etlorc/hainiuetlcategories10_orc/month=${yymm}/day=${dd}
  57. /usr/local/hadoop/bin/hadoop fs -cp hdfs://ns1/user/suyuan09/hainiuetl/orcspider/${yymm}/${dd}/etlAvro2Orc_spider_${mmdd}/part-*  hdfs://ns1/user/suyuan09/etlorc/hainiuetlspider_orc/month=${yymm}/day=${dd}
  58. /usr/local/hadoop/bin/hadoop fs -cp hdfs://ns1/user/suyuan09/hainiuetl/orcindex5/${yymm}/${dd}/etlAvro2Orc_index5_${mmdd}/part-*  hdfs://ns1/user/suyuan09/etlorc/hainiuetlindex5_orc/month=${yymm}/day=${dd}
  59. /usr/local/hadoop/bin/hadoop fs -cp hdfs://ns1/user/suyuan09/hainiuetl/orcip/${yymm}/${dd}/etlAvro2Orc_ip_${mmdd}/part-*  hdfs://ns1/user/suyuan09/etlorc/hainiuetlip_orc/month=${yymm}/day=${dd}
  64. #自动从hive到mysql脚本
  65. #hive2data.sh
  66.  
  67. /usr/local/hive/bin/hive  -e "use suyuan09;select t.top,t.num from(select top,count(*) num from hainiuetlindex5_orc group by top) t  sort by t.num desc limit 5;" >  /home/suyuan09/etl/hainiu/orc2mysql/myindex5${yymmdd}
  68. /usr/local/hive/bin/hive  -e "use suyuan09;select t.top,t.num from(select top,count(*) num from hainiuetltopics10_orc group by top) t    sort by t.num desc limit 10;" >    /home/suyuan09/etl/hainiu/orc2mysql/mytopics10${yymmdd}
  69. /usr/local/hive/bin/hive  -e "use suyuan09;select t.top,t.num from(select top,count(*) num from hainiuetlcategories10_orc  group by top) t  sort by t.num desc limit 10;" >   /home/suyuan09/etl/hainiu/orc2mysql/mycategories10${yymmdd}
  70. /usr/local/hive/bin/hive  -e "use suyuan09;select t.uip,t.num from(select uip,count(*) num from hainiuetlip_orc  group by uip) t  sort by t.num desc;" >   /home/suyuan09/etl/hainiu/orc2mysql/myip${yymmdd}
  71. /usr/local/hive/bin/hive  -e "use suyuan09;select t.usagent,t.num from(select usagent,count(*) num  from hainiuetlspider_orc  group by usagent) t   sort by t.num desc;" >  /home/suyuan09/etl/hainiu/orc2mysql/myspider${yymmdd}
  74. #data->mysql脚本
  75. #data2mysql.sh
  77. #mysql -h 172.33.101.123 -P 3306 -u tony -pYourPassword -D YourDbName <<EOF
  78. /bin/mysql -h192.168.88.195 -p3306 -uhainiu -p12345678 -Dhainiutest <<EOF
  80. LOAD DATA LOCAL INFILE "/home/suyuan09/etl/hainiu/orc2mysql/mytopics10${yymmdd}" INTO TABLE suyuan09_etl_orctopics10mysql FIELDS TERMINATED BY '\t';
  81. LOAD DATA LOCAL INFILE "/home/suyuan09/etl/hainiu/orc2mysql/mycategories10${yymmdd}" INTO TABLE suyuan09_etl_orccategories10mysql FIELDS TERMINATED BY '\t';
  82. LOAD DATA LOCAL INFILE "/home/suyuan09/etl/hainiu/orc2mysql/myindex5${yymmdd}" INTO TABLE suyuan09_etl_orcindex5mysql FIELDS TERMINATED BY '\t';
  83. LOAD DATA LOCAL INFILE "/home/suyuan09/etl/hainiu/orc2mysql/myspider${yymmdd}" INTO TABLE suyuan09_etl_orcspidermysql FIELDS TERMINATED BY '\t';
  84. LOAD DATA LOCAL INFILE "/home/suyuan09/etl/hainiu/orc2mysql/myip${yymmdd}" INTO TABLE suyuan09_etl_orcipmysql FIELDS TERMINATED BY '\t';
  86. EOF

报表展示

 

其中 mysql没有自带排序函数,自己写一个

 

热力图参考之前我之前的博客 https://www.cnblogs.com/symkmk123/p/9309322.html 其中之前是用 c# 写的,这里用java + spring 改写一下

思路看之前的博客这里放代码

经纬度转换类:LngAndLatUtil

  1. package suyuan.web;
  3. import java.io.BufferedReader;
  4. import java.io.IOException;
  5. import java.io.InputStreamReader;
  6. import java.io.UnsupportedEncodingException;
  7. import java.net.MalformedURLException;
  8. import java.net.URL;
  9. import java.net.URLConnection;
  11. public class LngAndLatUtil 
  12. {
  13.     public Object[] getCoordinate(String addr) throws IOException
  14.     {
  15.         String lng = null;// 经度
  16.         String lat = null;// 纬度
  17.         String address = null;
  18.         try
  19.         {
  20.             address = java.net.URLEncoder.encode(addr, "UTF-8");
  21.         } catch (UnsupportedEncodingException e1)
  22.         {
  23.             e1.printStackTrace();
  24.         }
  25.         String key = "你的秘钥";
  26.         String url = String.format("http://api.map.baidu.com/geocoder?address=%s&output=json&key=%s", address, key);
  27.         URL myURL = null;
  28.         URLConnection httpsConn = null;
  29.         try
  30.         {
  31.             myURL = new URL(url);
  32.         } catch (MalformedURLException e)
  33.         {
  34.            e.printStackTrace();
  35.         }
  36.         InputStreamReader insr = null;
  37.         BufferedReader br = null;
  38.         try
  39.         {
  40.             httpsConn = (URLConnection) myURL.openConnection();// 不使用代理
  41.             if (httpsConn != null)
  42.             {
  43.                 insr = new InputStreamReader(httpsConn.getInputStream(), "UTF-8");
  44.                 br = new BufferedReader(insr);
  45.                 String data = null;
  46.                 int count = 1;
  47.                 while ((data = br.readLine()) != null)
  48.                 {
  49.                     if (count == 5)
  50.                     {
  51.                         try{
  52.                         lng = (String) data.subSequence(data.indexOf(":") + 1, data.indexOf(","));// 经度
  54.                         count++;
  55.                         }
  56.                         catch(StringIndexOutOfBoundsException e)
  57.                         {
  58.                             e.printStackTrace();
  59.                         }
  60.                     } else if (count == 6)
  61.                     {
  62.                         lat = data.substring(data.indexOf(":") + 1);// 纬度
  63.                         count++;
  64.                     } else
  65.                     {
  66.                        count++;
  67.                    }
  68.                 }
  69.             }
  70.         } catch (IOException e)
  71.         {
  72.             e.printStackTrace();
  73.         } finally
  74.         {
  75.             if (insr != null)
  76.             {
  77.                 insr.close();
  78.             }
  79.             if (br != null)
  80.             {
  81.                 br.close();
  82.             }
  83.         }
  84.         return new Object[] { lng, lat };
  85.     }
  86. }

IPDTO:(数据库映射类)

 

  1. package suyuan.entity;
  3. public class IPDTO
  4. {
  5.     public String top;
  6.     
  7.     public Integer num;
  9.     public String getTop()
  10.     {
  11.         return top;
  12.     }
  14.     public void setTop(String top)
  15.     {
  16.         this.top = top;
  17.     }
  19.     public Integer getNum()
  20.     {
  21.         return num;
  22.     }
  24.     public void setNum(Integer num)
  25.     {
  26.         this.num = num;
  27.     }
  28.     
  29.     
  30. }

IP:(热力图json类)

  1. package suyuan.entity;
  3. public class IP
  4. {
  5.      public String lng ;
  7.      public String lat ;
  9.      public int count ;
  11.     public String getLng()
  12.     {
  13.         return lng;
  14.     }
  16.     public void setLng(String lng)
  17.     {
  18.         this.lng = lng;
  19.     }
  21.     public String getLat()
  22.     {
  23.         return lat;
  24.     }
  26.     public void setLat(String lat)
  27.     {
  28.         this.lat = lat;
  29.     }
  31.     public int getCount()
  32.     {
  33.         return count;
  34.     }
  36.     public void setCount(int count)
  37.     {
  38.         this.count = count;
  39.     }
  40.      
  41.      
  42. }

DAO层转换方法

  1. public List<IP> getip() throws SQLException
  2.     {
  3.         List<IPDTO>  ipdto = null;
  4.         List<IP> ipList=new ArrayList<IP>();
  5.         // 编写SQL语句
  6.         String sql = "SELECT top,num FROM `suyuan09_etl_orcipmysql`";
  7.         // 占位符赋值?
  8.         // 执行
  9.         
  10.         ipdto = qr.query(sql, new BeanListHandler<IPDTO>(IPDTO.class));
  11.         
  12.         for(IPDTO ips: ipdto)
  13.         {
  14.             IP ip=new IP();
  15.             Integer num = ips.getNum();
  16.             String top = ips.getTop();
  17.             // 封装
  18.             LngAndLatUtil getLatAndLngByBaidu = new LngAndLatUtil();
  19.             Object[] o = null;
  20.             try
  21.             {
  22.                 o = getLatAndLngByBaidu.getCoordinate(top);
  23.             } catch (IOException e)
  24.             {
  25.                 // TODO Auto-generated catch block
  26.                 e.printStackTrace();
  27.             }
  29.             ip.setLng(String.valueOf(o[0]));
  30.             ip.setLat(String.valueOf(o[1]));
  31.             ip.setCount(num);
  32.             ipList.add(ip);
  33.         }
  34.         
  35.         // 返回
  36.         return ipList;
  37.     }

控制器调用返回json:

  1.     @RequestMapping("/getip")
  2.     public @ResponseBody List<IP> getip()throws Exception{
  3.         return studentService.getip();
  4.         
  5.     }

jsp页面显示:

  1. <%@ page language="java" import="java.util.*" pageEncoding="UTF-8"%>
  2. <%@ taglib uri="http://java.sun.com/jsp/jstl/core" prefix="c"%>
  3. <%@ taglib uri="http://java.sun.com/jsp/jstl/fmt" prefix="fmt"%>
  4. <!DOCTYPE HTML>
  5.  
  6. <html>
  7. <head >
  8.     <meta charset="UTF-8">
  9.    
  10.     <script type="text/javascript" src="http://api.map.baidu.com/api?v=2.0&ak=1fReIR62vFbOc2vgrBnRAGyUtLkgoIIH"></script>
  11.     <script type="text/javascript" src="http://api.map.baidu.com/library/Heatmap/2.0/src/Heatmap_min.js"></script>
  12.    <script type="text/javascript"
  13.     src="${pageContext.request.contextPath}/js/jquery-1.9.1.js"></script>
  14.  
  15.     <title></title>
  16.      <style type="text/css">
  17.         ul,li{list-style: none;margin:0;padding:0;float:left;}
  18.         html{height:100%}
  19.         body{height:100%;margin:0px;padding:0px;font-family:"微软雅黑";}
  20.         #container{height:700px;width:100%;}
  21.         #r-result{width:100%;}
  22.     </style>    
  23. </head>
  24. <body>
  25.     <form id="form1" runat="server">
  26.     <div>
  27.         <div id="container"></div>
  28.         <div id="r-result">
  29.             <input type="button"  onclick="openHeatmap();" value="显示热力图"/><input type="button"  onclick="closeHeatmap();" value="关闭热力图"/>
  30.         </div>
  31.     </div>
  32.     </form>
  33. </body>
  34. </html>
  35.  
  36. <script type="text/javascript">
  37.     var map = new BMap.Map("container");          // 创建地图实例
  39.     var point = new BMap.Point(118.906886, 31.895532);
  40.     map.centerAndZoom(point, 15);             // 初始化地图，设置中心点坐标和地图级别
  41.     map.enableScrollWheelZoom(); // 允许滚轮缩放
  42.  
  43.  
  44.  
  45.     if (!isSupportCanvas()) {
  46.         alert('热力图目前只支持有canvas支持的浏览器,您所使用的浏览器不能使用热力图功能~')
  47.     }
  48.  
  49.    
  50.     heatmapOverlay = new BMapLib.HeatmapOverlay({ "radius": 20 });
  51.     map.addOverlay(heatmapOverlay);
  52.     heatmapOverlay.setDataSet({ data: function () {
  54.         var serie = [];
  55.         $.ajax({
  56.             url: "${pageContext.request.contextPath}/getip",
  57.             
  58.             dataType: "json",
  59.             async: false,
  60.             success: function (dataJson) {
  61.                 for (var i = 0; i < dataJson.length; i++) {
  62.                     var item = {
  63.                         //name: res.data.titleList7[i],
  64.                         //value: randomData()
  65.                         lat: dataJson[i].lat,
  66.                         lng: dataJson[i].lng,
  67.                         count: dataJson[i].count
  68.                     };
  69.                     serie.push(item);
  70.                 }
  71.             }
  72.         });
  75.         return serie;
  76.     } (), max: 100 });
  77.     //是否显示热力图
  78.     function openHeatmap() {
  79.         heatmapOverlay.show();
  80.     }
  81.     function closeHeatmap() {
  82.         heatmapOverlay.hide();
  83.     }
  84.     closeHeatmap();
  85.     function setGradient() {
  86.         /*格式如下所示:
  87.         {
  88.         0:'rgb(102, 255, 0)',
  89.         .5:'rgb(255, 170, 0)',
  90.         1:'rgb(255, 0, 0)'
  91.         }*/
  92.         var gradient = {};
  93.         var colors = document.querySelectorAll("input[type='color']");
  94.         colors = [].slice.call(colors, 0);
  95.         colors.forEach(function (ele) {
  96.             gradient[ele.getAttribute("data-key")] = ele.value;
  97.         });
  98.         heatmapOverlay.setOptions({ "gradient": gradient });
  99.     }
  100.     //判断浏览区是否支持canvas
  101.     function isSupportCanvas() {
  102.         var elem = document.createElement('canvas');
  103.         return !!(elem.getContext && elem.getContext('2d'));
  104.     }
  105. </script>        
  106.

 

图表也参考我之前的博客 https://www.cnblogs.com/symkmk123/p/9010514.html

 

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Hadoop热门文章

  • Hive中的去重 ：distinct,group by与ROW_Number()窗口函数
  • HBASE的安装过程及运行HBASE程序的需要配置的内容
  • hadoop--Unable to load native-hadoop library for your platform解决方法
  • Hive sampling 语法之TABLESAMPLE用法理解
  • idea项目运行时报错提示Could not locate executable null\bin\winutils.exe in the Hadoop binaries
  • hadoop 无法访问50070端口问题解决
  • Hadoop运行时遇到java.io.FileNotFoundException错误的解决方法
  • KUDU数据导入尝试一：TextFile数据导入Hive,Hive数据导入KUDU
  • Hive安装
  • Hadoop 学习笔记 生态

    • Hadoop推荐文章

  • 基于Hive的大数据分析系统
  • HBase 在统一内容平台业务的优化实践
  • hive窗口分析函数使用详解系列一
  • 两台实体机器4个虚拟机节点的Hadoop集群搭建（Ubuntu版）
  • 在阿里云和腾讯云的轻量应用服务器上搭建Hadoop集群
  • Haproxy搭建 Web 群集实现负载均衡
  • HBase Compaction 原理与线上调优实践
  • Hive执行计划之hive依赖及权限查询和常见使用场景
  • 工良出品：包教会，Hadoop、Hive 搭建部署简易教程
  • Centos7系统编译Hadoop3.3.4

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号