zoukankan      html  css  js  c++  java
  • spark hive 结合处理 把多行变成多列

    原数据格式 :

    gid       id      score

    a1        1       90

    a1        2      80

    a1       3      79

    a1       4      80

    a2      1       79

    a2      3       89

    a3      2        45

    a3     4        57

    a4     3          56

    a5     3            89

    ......

    要把数据变成:

    gid    id_1    id_2      id_3     id_4   score 

    a1     90    80    79   80

    a2   79      0             89         0

    a3        0        45           0         57

    .......

    数据量很大,上百亿条,所以用hive与spark结合的形式,在hive中见表,先把每个gid 的不同id都合并成一行,然后输入spark中进行分割处理成多列。

    一、hive先将每个gid的多行数据合并成一行

    create table a_id_score_concat as
    select a.gid,concat_ws(',',collect_set(concat(a.model_id,'=',a.score))) as score   #把model_id  与score用等号合并起来,便于后面拆开后匹配,也能解决每个gid的id个数不等的问题
    from a_id_score a group by a.gid;

    输出形式:

    a1   1=90,2=80,3=79,4=80

    a2    1=79,3=89

    二、用spark的函数,把合并的几行再分成多列,本质是python函数

    from pyspark.sql import SQLContext,Row

    from pyspark.mllib.regression import LabeledPoint
    from pyspark import SparkContext, SparkConf
    from pyspark.sql import HiveContext
    from pyspark.mllib.linalg import SparseVector, DenseVector

    df2=sqlContext.sql("select * from a_id_score_concat")  

    def splits(lt1):
          lt0=['null','10101','10102','10103','10105','10106','10121','10122','10123','10125','10126','10201','10221']  #id列表

            #需要合并的id,加一个null是为了跟out的列表长度一致,后面索引好匹配。
          out=['na','0','0','0','0','0','0','0','0','0','0','0','0']  #拆分出来的矩阵格式  列数据初始化,没有的id位置默认为0
          gid=lt1[0]
          sco=lt1[1].split(",")   #取出合并的score
          out[0]=gid
          for i in sco:
             s1=i.split("=")   #把每个等式拆开
             index1=lt0.index(s1[0])   #找索引
             out[index1]=s1[1]   #在索引对应列放入数据
          return out

    df3=sqlContext.createDataFrame(df2.map(splits,["gid",'m10101','m10102','m10103','m10105','m10106','m10121','m10122','m10123','m10125','m10126','m10201','m10221'])


    #保存dataframe,数据输出
    df3.saveAsTable(tableName="id_scores",source="parquet",mode="overwrite")

    hive中建表保存:

    hadoop fs -mv hdfs:./xiaofei_model_installed_pkgs hdfs:./zhangb.db

    # 在hive中建表语句
    create external table aaaaaa (
    pkg string,cnt01 bigint,cnt11 bigint,xsb double,chi2 double,gain double,iv double,rank1 int ,rank2 int ,rank3 int ,rank int )
    ROW FORMAT SERDE
    'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
    STORED AS INPUTFORMAT
    'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat'
    OUTPUTFORMAT
    'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
    LOCATION
    'hdfs:./zhangb.db/id_scores';

  • 相关阅读:
    python中的面向对象编程
    python global vs nonlocal (2)
    python3 nonlocal vs global
    poj蚂蚁问题
    C/C++ static vs global
    砝码问题
    Wythoff's game
    C++中的::operator new, ::operator delete
    客户信息表 自我汇总 待确认
    Oracle Savepoint
  • 原文地址:https://www.cnblogs.com/zhangbojiangfeng/p/6169923.html
Copyright © 2011-2022 走看看