直接上代码

package com.tophant

import java.text.SimpleDateFormat

import com.tophant.html.Util_html
import com.tophant.util.Util
import org.apache.commons.lang.{StringEscapeUtils, StringUtils}
import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.hadoop.hbase.client.Scan
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter
import org.apache.hadoop.hbase.util.{Base64, Bytes}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.slf4j.LoggerFactory

object Hbase2hdfs {
  def main(args: Array[String]): Unit = {
    val logger = LoggerFactory.getLogger(getClass)
    // 命令行参数个数必须为2
    if (args.length != 5) {
      logger.error("参数个数错误")
      logger.error("Usage: Classify <开始日期> <结束日期> <存储位置> <host>")
      System.exit(1)
    }

    // 获取命令行参数中的行为数据起止日期
    val tableName = args(0).trim  //hbase表的前缀
    val startDate = args(1).trim  //hbase表的后缀起始日期
    val endDate   = args(2).trim  //hbase表的后缀截至日期
    val path = args(3).trim
    val host = args(4).trim
    // 根据起止日志获取日期列表
    // 例如起止时间为20160118,20160120,那么日期列表为(20160118,20160119,20160120)
    val dateSet = getDateSet(startDate, endDate)

    val sparkConf = new SparkConf()
      .setAppName("hbase2hdfs")
    //      .set("spark.default.parallelism","12")
    val sc = new SparkContext(sparkConf)

    val conf = HBaseConfiguration.create()
    //设置zooKeeper集群地址，也可以通过将hbase-site.xml导入classpath，但是建议在程序里这样设置
    conf.set("hbase.zookeeper.quorum", "10.0.71.501,10.0.71.502,10.0.71.503")
    //设置zookeeper连接端口，默认2181
    conf.set("hbase.zookeeper.property.clientPort", "2181")

    //组装scan语句
    val scan=new Scan()
    scan.setCacheBlocks(false)
    scan.addFamily(Bytes.toBytes("rd"))
    scan.addColumn(Bytes.toBytes("rd"), Bytes.toBytes("host"))
    scan.addColumn("rd".getBytes, "url".getBytes)
    scan.addColumn("rd".getBytes, "http.request.method".getBytes)
    scan.addColumn("rd".getBytes, "http.request.header".getBytes)


    val hostFilter = new SingleColumnValueFilter(Bytes.toBytes("rd"),
      Bytes.toBytes("host"),
      CompareOp.EQUAL,
      Bytes.toBytes(host))
    scan.setFilter(hostFilter)
    import org.apache.hadoop.hbase.mapreduce.TableInputFormat
    import org.apache.hadoop.hbase.protobuf.ProtobufUtil
    val proto = ProtobufUtil.toScan(scan)
    val ScanToString = Base64.encodeBytes(proto.toByteArray)


    // 按照日期列表读出多个RDD存在一个Set中,再用SparkContext.union()合并成一个RDD
    var rddSet: Set[RDD[(org.apache.hadoop.hbase.io.ImmutableBytesWritable, org.apache.hadoop.hbase.client.Result)] ] = Set()
    dateSet.foreach(date => {
      conf.set(TableInputFormat.SCAN, ScanToString)
      conf.set(TableInputFormat.INPUT_TABLE, tableName + date) // 设置表名
      val bRdd: RDD[(org.apache.hadoop.hbase.io.ImmutableBytesWritable, org.apache.hadoop.hbase.client.Result)] =
        sc.newAPIHadoopRDD(conf, classOf[TableInputFormat],
        classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable],
        classOf[org.apache.hadoop.hbase.client.Result])
      rddSet += bRdd
    })

    val hBaseRDD = sc.union(rddSet.toSeq)

    val kvRDD = hBaseRDD.
      map { case (_, result) =>
        //通过列族和列名获取列
        var url = Bytes.toString(result.getValue("rd".getBytes, "url".getBytes))
        val host = Bytes.toString(result.getValue("rd".getBytes, "host".getBytes))
        var httpRequestHeaders = Bytes.toString(result.getValue("rd".getBytes, "http.request.header".getBytes))
        var httpResponseHeaders = Bytes.toString(result.getValue("rd".getBytes, "http.response.header".getBytes))

  

        val value = (url + "$$$" + httpRequestHeaders + "$$$" + httpResponseHeaders )

        ( host, value)
      }
    .saveAsTextFile(path)
  }

  def getDateSet(sDate:String, eDate:String): Set[String] = {
    // 定义要生成的日期列表
    var dateSet: Set[String] = Set()

    // 定义日期格式
    val sdf = new SimpleDateFormat("yyyyMMdd")

    // 按照上边定义的日期格式将起止时间转化成毫秒数
    val sDate_ms = sdf.parse(sDate).getTime
    val eDate_ms = sdf.parse(eDate).getTime

    // 计算一天的毫秒数用于后续迭代
    val day_ms = 24*60*60*1000

    // 循环生成日期列表
    var tm = sDate_ms
    while (tm <= eDate_ms) {
      val dateStr = sdf.format(tm)
      dateSet += dateStr
      tm = tm + day_ms
    }

    // 日期列表作为返回
    dateSet
  }
}

 

业务场景：在用户轨迹上增加筛选功能，实现根据标签1和2来筛选
解决思路：在原有数据上增加接收前端的数据的字段，然后在controller层处理字段的信息，然后分别增加条件进行判断实现筛选功能--只写下我修改的部分(此文章的目的就是为了在日常中查缺补漏，便于日后的复习和学习)
代码实现：
1.在dao层实现字段增加来接收前端的数据(和前端人员沟通确定参数的形式)
2.在实现层对条件2实现筛选if (!StringUtil.isEmpty(params.getType())){
String[] str2 = params.getType().split(",");
Set filter = new HashSet<>();
for (String type : str2) {
filter.add(type);}
query.addMustTerm(new AbstractTermV2.MustMultTerm("deviceOs",filter));
}
3.在controller增加筛选条件long total = 0;
SimpleQuery.SimpleResponse rumResponse = null;
//浏览器
if(!StringUtil.isEmpty(params.getTermType())) {
String n = "浏览器";
String[] str = params.getTermType().split(",");
Set termType = new HashSet<>();
for (String string : str) {
termType.add(string);
}
if (termType.contains(n)) {
rumResponse = queryRum(params, cluster.keySet());
total = rumResponse.getTotal();
}
}
rumResponse = queryRum(params, cluster.keySet());
total = rumResponse.getTotal();
在上面的业务实现过程中，所涉及的知识点是后端如何接收前端的数据，Java中使用.split()的方法,以及集合遍历的知识：后端接收前端数据(https://blog.csdn.net/justry_...)
1-1：后端参数是一个对象，且该参数前是以@RequestBody修饰的，那么前端传递json参数时，必须满足以下要求：
后端@RequestBody注解对应的类在将HTTP的输入流(含请求体)装配到目标类(即：@RequestBody后面的类)时，会根据json字符串中的key来匹配对应实体类的属性，如果匹配一致且json中的该key对应的值符合(或可转换为)实体类的对应属性的类型要求时,会调用实体类的setter方法将值赋给该属性。
1-2：json字符串中，如果value为""的话，后端对应属性如果是String类型的，那么接受到的就是""，如果是后端属性的类型是Integer、Double等类型，那么接收到的就是null。
1-3：json字符串中，如果value为null的话，后端对应收到的就是null
1-4：如果某个参数没有value的话，在传json字符串给后端时，要么干脆就不把该字段写到json字符串中；要么写value时， 必须有值，null  或""都行。千万不能有类似"stature":，这样的写法，如{"":,--->X
"":null,--->正确
"":""---->正确}
@RequestBody直接以String接收前端传过来的json数据：
后端对应的Controller：
使用PostMan测试：
2.遍历集合的方法
参考：https://www.jb51.net/article/...
3.split()的方法使用
使用此方法返回的是一个数组(以特定的格式分隔后的数组)
特殊情况有 * ^ : | . \
在使用这些为分隔条件是需注意必须使用转义符即\加以转义，然后如果使用多个分隔符则需要借助 | 符号，如二所示，但需要转义符的仍然要加上分隔符进行处理
参考：https://www.cnblogs.com/mingf...

    
例如以下表:

菜品表: foods




字段
说明



id
主键


title
菜品名称




菜品规格表: foods_specs




字段
说明



id
主键


title
规格名称


foods_id
菜品表ID


price
价格




菜品模型: FoodsModel


class FoodsModel extends Model
{
  
     // 省略
     /**
     * 规格
     * @return HasMany
     */
    public function specs()
    {
        return $this->hasMany(\App\Model\FoodsSpecsModel::class, 'foods_id', 'id');
    }
 
   // 省略
}

菜品规格模型: FoodsSpecsModel
class FoodsSpecsModel extends Model
{
    protected $table = 'foods_specs';
 
    protected function foods()
    {
        return $this->belongsTo('App\Model\FoodsModel', 'foods_id', 'id');
    }
}

根据菜品价格区间查询筛选菜品:
FoodsModel::with(['specs' => function($query){
            $query->select(['title', 'price']);
        }])
        ->whereHas('specs', function ($query) use ($params){
            // price_from,price_end对应价格区间, 可选参数
            if (isset($params['price_from']) && isset($params['price_end'])){
                $query->whereBetween('price', [$params['price_from'], $params['price_end']]);
            }
        })->select('title')->get();

在写join类型的语句的时候，最好使用join来代替直接用“，”分隔多个表。这样必须提供on条件才能进行查询。当然，join的条件一定要设置好，否则也会出现笛卡尔积的。 


比如tableA和TableB向关联的是ID字段 


如果用“,“分隔两个表的话语据应该如下 


select count(*) 

from tableA ,TableB 

where TableA.id=TableB.id 


如果把where 语句漏掉就出现笛卡尔积了 

select count(*) 

from tableA ,TableB 


同样，写成join的话，如果不指定on里的条件的话，语法是错的 

select count(*) 

from TableA join TableB 

on TableA.id=TableB.id 


但是如果再on里指定了一个错误的关联条件的话，也是会出现笛卡尔积的,比如： 

select count(*) 

from TableA join TableB 

on 1=1 


当然一般人不会闲得去把on条件设置成1=1,不过这是举个例子，就是为了说明on的筛选条件一定要设置好。

转载于:https://www.cnblogs.com/yhb199/archive/2008/06/11/1217572.html