плоская карта scala rdd для создания нескольких строк из одной строки для заполнения пробела в строке - PullRequest
Новая
       26

плоская карта scala rdd для создания нескольких строк из одной строки для заполнения пробела в строке

0 голосов
/ 06 января 2019

Я пытаюсь решить проблему, когда скажем, что человек занял деньги у кого-то, и тогда у нас есть вся операция по возврату этих денег в рассрочку без процентов. Здесь я хочу заполнить неоплаченные строки той же суммой, что и предыдущая сумма

Input 

name,date_of_borrow/return,Amount-Principal
Ashish,2018-03-01,20000
Ashish,2018-04-01,19000
Ashish,2018-05-01,18000
Ashish,2018-06-01,17000
Ashish,2018-07-01,16000
Ashish,2018-08-01,15000
Ashish,2018-12-01,14000
Ashish,2019-02-01,13000

Ожидаемый результат 

name,date_of_borrow/return,Amount-principal
Ashish,2018-03-01,20000
Ashish,2018-04-01,19000
Ashish,2018-05-01,18000
Ashish,2018-06-01,17000
Ashish,2018-07-01,16000
Ashish,2018-08-01,15000
****Ashish,2018-09-01,15000**  --- copy previous amount as installment not paid
**Ashish,2018-10-01,15000**
**Ashish,2018-11-01,15000****
Ashish,2018-12-01,14000
**Ashish,2018-01-01,14000**
Ashish,2019-02-01,13000

Я хочу написать это в Scala RDD 

val tr = spark.sparkContext.textFile("/tmp/data.txt")
val tr.map(x=>x.split(',')).map(x=>(x(0),(x(1),x(2)))).collect()
val sm= tr.map(x=>(x.split(',')(0),(x))).groupByKey().flatMap(rec=>{rec._2.toList.sortBy(x=>(-x.split(",")(2).toFloat)).zipWithIndex})
val part1 = sm.map(x=>((x._1.split(',')(0),x._2.toInt),(x._1.split(',')(1),x._1.split(',')(2))))
val part2 = sm.map(x=>((x._1.split(',')(0),x._2.toInt-1),(x._1.split(',')(1),x._1.split(',')(2))))
val data = part1.leftOuterJoin(part2).sortByKey()

/// Чтение данных и соединение со следующей строкой для имени на основе индекса 

val oo = data.map(x=>(x._1._1,x._2._1._1,x._2._1._2,x._2._2.getOrElse(x._2._1._1,0)))
val rr  = oo.map(x=>(x._1,x._2,x._3,x._4._1))  or     val oo = data.map(x=>(x._1._1,x._2._1._1,x._2._1._2,x._2._2.getOrElse(x._2._1._1,0)._1))

/// Отображение для получения окончательных данных как 

scala> oo.filter(x=>x._1=="Ashish").collect().foreach(println)
(Ashish,2018-03-01,20000,2018-04-01)
(Ashish,2018-04-01,19000,2018-05-01)
(Ashish,2018-05-01,18000,2018-06-01)
(Ashish,2018-06-01,17000,2018-07-01)
(Ashish,2018-07-01,16000,2018-08-01)
(Ashish,2018-08-01,15000,2018-12-01)
(Ashish,2018-12-01,14000,2019-02-01)
(Ashish,2019-02-01,13000,2019-02-01)

Теперь остальная часть задачи по поиску различий по датам и созданию строк flatMap 

val format = new java.text.SimpleDateFormat("yyyy-MM-dd")
format.format(new java.util.Date())  --test date
def generateDates(startdate: Date, enddate: Date): ListBuffer[String] ={
var dateList = new ListBuffer[String]()
var calendar = new GregorianCalendar()
calendar.setTime(startdate)
while (calendar.getTime().before(enddate)) {
dateList += (calendar.get(Calendar.YEAR)) + "-" + (calendar.get(Calendar.MONTH)+1) + "-" +  (calendar.get(Calendar.DAY_OF_MONTH)) 
calendar.add(Calendar.MONTH, 1)
}
if (dateList.isEmpty) { 
  dateList+= (calendar.get(Calendar.YEAR)) + "-" + (calendar.get(Calendar.MONTH)+1) + "-" +  (calendar.get(Calendar.DAY_OF_MONTH))}
println("\n" + dateList + "\n")
dateList
}

Вот где это идет не так, и я сталкиваюсь с трудностями в понимании или решении этого. Я получаю одну дополнительную строку за каждую последнюю дату, которая не должна прийти 

scala> oo.filter(x=>x._1=="Ashish").flatMap(pp=> {
     |     var allDates = new ListBuffer[(String,String,Integer)]()
     |     for (x <- generateDates(format.parse(pp._2),format.parse(pp._4))) {
     |     allDates += ((pp._1, x , pp._3.toInt))}
     |     allDates
     |     }).collect().foreach(println)


(Ashish,Thu Mar 01,1,2,20000)                                                   
(Ashish,Sun Apr 01 00:00:00 IST 2018,20000)   --- unwanted row and I dont know why wrong date format
(Ashish,Sun Apr 01,1,3,19000)
(Ashish,Tue May 01 00:00:00 IST 2018,19000)   --- unwanted row
(Ashish,Tue May 01,1,4,18000)
(Ashish,Fri Jun 01 00:00:00 IST 2018,18000)
(Ashish,Fri Jun 01,1,5,17000)
(Ashish,Sun Jul 01 00:00:00 IST 2018,17000)
(Ashish,Sun Jul 01,1,6,16000)
(Ashish,Wed Aug 01 00:00:00 IST 2018,16000)
(Ashish,Wed Aug 01,1,7,15000)
(Ashish,Sat Sep 01,1,8,15000)
(Ashish,Mon Oct 01,1,9,15000)
(Ashish,Thu Nov 01,1,10,15000)
(Ashish,Sat Dec 01 00:00:00 IST 2018,15000)
(Ashish,Sat Dec 01,1,11,14000)
(Ashish,Tue Jan 01,1,0,14000)
(Ashish,Fri Feb 01 00:00:00 IST 2019,14000)
(Ashish,Fri Feb 01 00:00:00 IST 2019,13000)

Я полностью согласен с тем, что это может быть плохим способом написания кода, но кто-то может помочь в его понимании. потому что мне нужно понять, где я ошибаюсь, а также мне нужно знать лучший способ сделать это

Я вижу, что функция даты работает нормально. 

scala> for(x<-generateDates(format.parse("2018-01-01"),format.parse("2018-11-01")))
     | {
     | println("\n" + x + "\n")
     | }

ListBuffer(2018-1-1, 2018-2-1, 2018-3-1, 2018-4-1, 2018-5-1, 2018-6-1, 2018-7-1, 2018-8-1, 2018-9-1, 2018-10-1)


2018-1-1


2018-2-1


2018-3-1


2018-4-1


2018-5-1


2018-6-1


2018-7-1


2018-8-1


2018-9-1


2018-10-1

1 Ответ

0 голосов
/ 30 января 2019

Я все еще работаю над поиском причины, по которой этот код не работал, но я кодировал его другим способом, который дает мне правильный результат 

import org.apache.spark.{ SparkConf, SparkContext }
import org.apache.spark.sql.functions.broadcast
import org.apache.spark.sql.types._
import org.apache.spark.sql._
import org.apache.spark.sql.functions._
import scala.collection.mutable.ListBuffer
import java.util.{GregorianCalendar, Date}
import java.util.Calendar
val ipt = spark.read.format("com.databricks.spark.csv").option("header","true").option("inferchema","true").load("/tmp/data.csv")
val sm  = ipt.rdd.map(x=>(x(0).toString(),(x.toString().replace("[","").replace("]","")))).groupByKey().flatMap(rec=>{rec._2.toList.sortBy(x=>(-x(2).toFloat)).zipWithIndex})
val part1 = sm.map(x=>((x._1.split(',')(0),x._2.toInt),(x._1.split(',')(1),x._1.split(',')(2))))
val part2 = sm.map(x=>((x._1.split(',')(0),x._2.toInt-1),(x._1.split(',')(1),x._1.split(',')(2))))
val data = part1.leftOuterJoin(part2).sortByKey()
val oo = data.map(x=>(x._1._1,x._2._1._1,x._2._1._2,x._2._2.getOrElse(x._2._1._1,0)))
val oo = data.map(x=>(x._1._1,x._2._1._1,x._2._1._2,x._2._2.getOrElse(x._2._1._1,0)._1))
val format = new java.text.SimpleDateFormat("yyyy-MM-dd")
format.format(new java.util.Date())  --test date
def generateDates(startdate: Date, enddate: Date): ListBuffer[String] ={
var dateList = new ListBuffer[String]()
var calendar = new GregorianCalendar()
calendar.setTime(startdate)
while (calendar.getTime().before(enddate)) {
dateList += (calendar.get(Calendar.YEAR)) + "-" + (calendar.get(Calendar.MONTH)+1) + "-" +  (calendar.get(Calendar.DAY_OF_MONTH)) 
calendar.add(Calendar.MONTH, 1)
}
if (dateList.isEmpty) { 
  dateList+= (calendar.get(Calendar.YEAR)) + "-" + (calendar.get(Calendar.MONTH)+1) + "-" +  (calendar.get(Calendar.DAY_OF_MONTH))}
println("\n" + dateList + "\n")
dateList
}
oo.flatMap(pp=> {
var allDates = new ListBuffer[(String,String,Integer)]()
for (x <- generateDates(format.parse(pp._2),format.parse(pp._4))) {
allDates += ((pp._1, x , pp._3.toInt))}
allDates
}).collect().foreach(println)





(Ashish,2018-3-1,20000)
(Ashish,2018-4-1,19000)
(Ashish,2018-5-1,18000)
(Ashish,2018-6-1,17000)
(Ashish,2018-7-1,16000)
(Ashish,2018-8-1,15000)
(Ashish,2018-9-1,15000)
(Ashish,2018-10-1,15000)
(Ashish,2018-11-1,15000)
(Ashish,2018-12-1,14000)
(Ashish,2019-1-1,14000)
(Ashish,2019-2-1,13000)
...