在工作中经常使用Spark处理 非标准化的数据，将数据处理成转换为 DataFrame对象。并从DataFrame对象中创建视图，构建数据表，以数据表的形式保存最终数据。
关键步骤：a、通过case class定义schema。b、解析数据，得到包含case class对象的RDD。 c、将RDD转换为DataFrame对象。d、指定数据库，新建数据表。通过DataFrame对象创建视图，讲视图数据插入到数据表中。

知识点

1、Spark如何支持Hive操作

2.x版本

  1. import org.apache.spark.sql.SparkSession
  3. //case class 
  4. case class PredictResult(userid:String, item_id:String,
  5.                          pctr:Float, pcvr:Float, pay_label:Int,
  6.                          click_label:Int)
  9. // 新建会话
  10. val sparkSession = SparkSession.builder()
  11.       .appName("mlx_feature_process")
  12.       .config("hive.exec.dynamic.partition", true)
  13.       .config("hive.exec.dynamic.partition.mode", "nonstrict")
  14.       .enableHiveSupport()
  15.       .getOrCreate()
  17. // 读取数据，解析数据。构建case class 的RDD。
  18. val sc = sparkSession.sparkContext
  19. val predictSample = sc.textFile(...).map(r=>{PredictResult(r)}) 
  21. // RDD 转化为 DataFrame对象，并创建视图
  22. import sparkSession.implicits._
  23. predictSample.toDF().createOrReplaceTempView("predict_table")
  25. // 数据表名
  26. val  tableName = "predict_sample_result_"+model
  27. // 指定使用的数据库
  28. sparkSession.sql("use ba_xxx")
  29. // 执行建表操作
  30. sparkSession.sql(createTableSql(tableName))
  32. // 执行数据建表操作
  33. sparkSession.sql(
  34.       s"""
  35.          |alter table $tableName drop if exists partition(dt=$date)
  36.          |""".stripMargin)
  37. sparkSession.sql(
  38.       s"""
  39.          |insert into $tableName partition(dt=$date)
  40.          |select userid,item_id,item_type,
  41.          |pctr,pcvr,pay_label,click_label, ts
  42.          |from predict_table""".stripMargin)

建表语句：

  1. def createTableSql(tableName:String): String ={
  2.     val sqlStr1 =
  3.       s"""
  4.         |CREATE TABLE IF NOT EXISTS $tableName(
  5.         |      `userid` string comment 'userid',
  6.         |      `item_id` string comment 'item_id',
  7.         |      `item_type` string comment 'item_type',
  8.         |      `pctr` float comment '模型预测ctr',
  9.         |      `pcvr` float comment '模型预测cvr',
  10.         |      `pay_label` int comment '支付标签，-1表示无效',
  11.         |      `click_label` int comment '点击标签',
  12.         |      `ts` string comment '时间戳'
  13.         |) COMMENT '推荐报表'
  14.         |PARTITIONED BY(dt string COMMENT "yyyyMMdd,日期分区")
  15.         |STORED AS ORC
  16.             """.stripMargin
  17.       sqlStr1
  18.   }

2、DataSet、DataFrame如何转换成数据表

使用隐式强转

  1. // 转换，创建视图
  2. import sparkSession.implicits._
  3. predictSample.toDF().createOrReplaceTempView("predict_table")

3、Spark如何支持HiveSql操作

和正常Sql操作没有任务区别

  1. 使用 sparkSession.sql(sql语句) 执行sql任务。

FAQ

1、为什么使用数据库表，而不使用HDFS存储原始数据？
格式化的标准数据，便于管理、维护。

参考

1、RDD对象无法调用toDF的解决方法