1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
| import java.text.SimpleDateFormat
import java.util.{Calendar, Date}
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.{SparkConf, SparkContext, mllib}
import org.apache.spark.mllib.linalg.{Vector, Vectors}
import org.apache.spark.mllib.recommendation.ALS
import com.soundcloud.lsh.{Cosine, Lsh, NearestNeighbours}
import org.apache.spark.mllib.linalg.distributed.{IndexedRow, IndexedRowMatrix}
import org.apache.spark.storage.StorageLevel
import org.apache.spark.mllib.recommendation.Rating
import scala.collection.mutable.ArrayBuffer
object CosineLSHJoinSpark {
def main(args: Array[String]){
val conf = new SparkConf().setAppName("CosineLSHJoinSpark")
val sc = new SparkContext(conf)
val sqlContext = new org.apache.spark.sql.SQLContext(sc)
val df = sqlContext.read.format("orc").load(getLastNDaysPath(15))
df.show
val (ratings,productMap) = datatransform_formkpixel(df)
ratings.take(20).foreach(println)
val rank = 10
val numIterations = 10
val model = ALS.train(ratings, rank, numIterations, 0.02)
val vectorArray: ArrayBuffer[IndexedRow] = new ArrayBuffer[IndexedRow]()
val fdata=model.productFeatures.collect()
for(i <- 0 until fdata.length) {
if(fdata(i)!=null && fdata(i)._1!=null && fdata(i)._2!=null) {
val t=fdata(i)
val ir=IndexedRow(i,Vectors.dense(t._2))
vectorArray.append(ir)
}
}
val idxrows = sc.parallelize(vectorArray)
val idxmat: IndexedRowMatrix = new IndexedRowMatrix(idxrows)
val lsh = new Lsh(
minCosineSimilarity = 0.1,
dimensions = 20,
numNeighbours = 10,
numPermutations = 2
)
val similariyMatrix = lsh.join(idxmat)
val orderTable=similariyMatrix.entries.groupBy(tup => tup.i).flatMap(tup =>{
tup._2.toList.sortWith((a,b) =>a.value>b.value)
})
val results = orderTable.map {
entry =>
"%s %s %.6f".format(productMap(entry.i), productMap(entry.j), entry.value)
}
results.take(20).foreach(println)
results.saveAsTextFile("this is save path ")
}
def datatransform(df:DataFrame) ={
val r = df.rdd
val affData = r.flatMap(row => {
val result: ArrayBuffer[(String, String)] = new ArrayBuffer[(String, String)]()
val ifa: String = if (row(0) != null) row.getString(0) else null
val bundle: Array[String] = if (row(1) != null) row.getSeq[String](1).toArray[String] else null
if (bundle != null && bundle.length > 1) {
bundle.foreach(b => {
result.append((ifa, b.trim))
})
}
result
}).filter(x => x._1 != null && x._2 != null )
val stringData=affData.map(x =>(x._1,x._2,1))
val ifaname = stringData.map(_._1).distinct.sortBy(x => x).zipWithIndex.collectAsMap
val products = stringData.map(_._2).distinct.sortBy(x => x).zipWithIndex.collectAsMap
val data_rating=stringData.map(r => Rating(ifaname(r._1).toInt,products(r._2).toInt,r._3))
val reproducts=products.map(line=>(line._2,line._1))
(data_rating,reproducts)
}
def datatransform_formkpixel(df:DataFrame) ={
val r = df.rdd
val affData = r.flatMap(row => {
val result: ArrayBuffer[(String, String)] = new ArrayBuffer[(String, String)]()
val ifa: String = if (row(8) != null) row.getString(8) else null
val content = if (row(16) != null) row.getString(16) else null
val content_ids: Array[String] = if (content != null) content.split(",") else null
if (content_ids != null && content_ids.length > 1) {
content_ids.foreach(b => {
result.append((ifa, b.trim))
})
}
result
}).filter(x => x!=null && x._1 != null && x._2 != null)
val stringData=affData.map(x =>(x._2,x._1,1))
val ifaname = stringData.map(_._1).distinct.sortBy(x => x).zipWithIndex.collectAsMap
val products = stringData.map(_._2).distinct.sortBy(x => x).zipWithIndex.collectAsMap
val data_rating=stringData.map(r => Rating(ifaname(r._1).toInt,products(r._2).toInt,r._3))
val reproducts=products.map(line=>(line._2,line._1))
(data_rating,reproducts)
}
def getLastNDaysPath(days: Int ): String = {
val dateFormat: SimpleDateFormat = new SimpleDateFormat( "yyyyMMdd" )
val date=dateFormat.format(new Date())
val dateF: Date = dateFormat.parse(date)
val cal: Calendar = Calendar.getInstance()
cal.setTime(dateF)
val dateArr = new ArrayBuffer[String]()
var i = 1
while ( i <= days ) {
cal.add( Calendar.DATE, -i )
dateArr.append(dateFormat.format( cal.getTime() ))
i = i + 1
cal.setTime(dateF)
}
val path = "this is path prefix/{"+ dateArr.mkString(",") +"}/*/*"
println(path)
path
}
}
|
