天池新人実戦試合の[オフライン]-初体験-Spark処理

前に書く
このようなショッピングカートを直接提出する方法はアルゴリズムにかかわらず、その後、機械学習を導入する関連アルゴリズムを徐々に学習し、結果をよりよく予測します.2019-08-07までランキング77/1111

package src.main.scala.com.csylh.august.tianchi.dataclearer

import org.apache.spark.sql.{SaveMode, SparkSession}

/**
  *     
  */
import org.apache.spark.sql.functions.split


/**
  * Description: TODO
  *
  * @Author:   36
  * @Date: 2019-08-01 10:57
  */
object SourceDataETLApp {
  def main(args: Array[String]): Unit = {
    val localMasterURL = "local[2]"
    val clusterMasterRL = ""

    //   SparkSession  
    val spark = SparkSession.builder()
      .master(localMasterURL)
      .appName("SourceDataETLApp")
      .getOrCreate()

    val itemData = "/Users/liuge36/Desktop/fresh_comp_offline/tianchi_fresh_comp_train_item.csv"
    val userData = "/Users/liuge36/Desktop/fresh_comp_offline/tianchi_fresh_comp_train_user.csv"

    /**
      *    ： 620918
      * |  item_id|item_geohash|item_category|
      * +---------+------------+-------------+
      * |100002303|        null|         3368|
      *
      */
    //       （P）
    val train_item =spark.read.option("header",true).csv(itemData)
    //           
//    println(train_item.count())
    //    10       
//    train_item.show(10)

    /**
      *
      *    ： 23 291 027
      * +--------+---------+-------------+------------+-------------+-------------+
      * | user_id|  item_id|behavior_type|user_geohash|item_category|         time|
      * +--------+---------+-------------+------------+-------------+-------------+
      * |10001082|285259775|            1|     97lk14c|         4076|2014-12-08 18|
      */
      //         
    val train_user =spark.read.option("header",true).csv(userData)
    //         
//    println(train_user.count())
    //    10     
//    train_user.show(10)
    //          
//    train_user.select("time","behavior_type").show()

    //    behavior_type==3，         659437
    val resultData = train_user.filter("behavior_type == 3")
    //     
    import spark.implicits._

    //    12 18       18487 ,     user_id  item_id   

    val saveData = resultData.withColumn("_tmp", split(resultData.col("time"), " "))
      .select($"_tmp".getItem(0).as("t1"), $"user_id", $"item_id")
      .filter("t1 == '2014-12-18' ")
      .drop("_tmp", "t1")

    //     
    saveData
      .coalesce(1)
      .write
      .option("header", "true")
      .mode(SaveMode.Overwrite)

      .csv("/Users/liuge36/Desktop/fresh_comp_offline/2")


    spark.stop()


  }

}