【Use Case】Predicting Geographic Population using Genome Variants and K-Means

1. 9) Prepare data for K-means clustering
   
2. >
   
3. 
   
4. import org.apache.spark.mllib.linalg.{Vector=>MLVector, Vectors}
   
5. 
   
6. val sampleToData: RDD[(String, (Int, Double))] = finalGts.map { g => (g.getSampleId.toString, ((variantId(g).hashCode), altAsDouble(g))) }
   
7. 
   
8. // group our data by sample
   
9. val groupedSampleToData = sampleToData.groupByKey
   
10. 
    
11. 
    
12. // make an MLVector for each sample, which contains the variants in the exact same order
    
13. def makeSortedVector(g: Iterable[(Int,Double)]): MLVector = Vectors.dense( g.toArray.sortBy(_._1).map(_._2) )
    
14. 
    
15. val dataPerSampleId:RDD[(String, MLVector)] =
    
16.     groupedSampleToData.mapValues { it =>
    
17.         makeSortedVector(it)
    
18.     }

复制代码

1. 10) Run K-means clustering to build model
   
2. >
   
3. 
   
4. import org.apache.spark.mllib.clustering.{KMeans,KMeansModel}
   
5. 
   
6. // Cluster the data into three classes using KMeans
   
7. val numClusters = 3
   
8. val numIterations = 20
   
9. val clusters:KMeansModel = KMeans.train(features, numClusters, numIterations)
   
10. 
    
11. // Evaluate clustering by computing Within Set Sum of Squared Errors
    
12. val WSSSE = clusters.computeCost(features)
    
13. println(s"Compute Cost: ${WSSSE}")

复制代码

1. 11) Predict populations, compute the confusion matrix.
   
2. >
   
3. 
   
4. // Create predictionRDD that utilizes clusters.predict method to output the model's predictions
   
5. val predictionRDD: RDD[(String, Int)] = dataPerSampleId.map(sd => {
   
6.     (sd._1, clusters.predict(sd._2))
   
7. })
   
8. 
   
9. // Convert to DataFrame to more easily query the data
   
10. val predictDF = predictionRDD.toDF("sample","prediction")
    
11. predictionRDD: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[2546] at map at <console>:105
    
12. predictDF: org.apache.spark.sql.DataFrame = [sample: string, prediction: int]
    
13. >
    
14. 
    
15. // Join back to the filterPanel to get the original label
    
16. val resultsDF =  filterPanel.join(predictDF, "sample")
    
17. display(resultsDF)

复制代码

1. 12) Visualize the clusters with a force graph on lightning-viz.
   
2. Create the input data vectors for the lightning-viz:
   
3. list of the group memberships (the populations)
   
4. list of the people (the sample IDs)
   
5. Nested list of the graph links, with each person linked to their predicted cluster.
   
6. >
   
7. 
   
8. %python
   
9. 
   
10. #prepare our data into a suitable format for our viz
    
11. 
    
12. from pyspark.sql.functions import rowNumber
    
13. from pyspark.sql.window import Window
    
14. 
    
15. #ensure that our data arrays come out in the same order
    
16. 
    
17. df = sqlContext.sql("select sample, popcode, prediction from final_results_table").coalesce(1)
    
18. w = Window().orderBy()
    
19. df = df.withColumn("rownumber", rowNumber().over(w))
    
20. 
    
21. pop = df.select("popcode", "rownumber").collect()
    
22. peeps = df.select("sample", "rownumber").collect()
    
23. preds= df.select("prediction", "rownumber").collect()
    
24. 
    
25. pop = [(str(x),str(y)) for (x,y) in pop]
    
26. peeps = [(str(x),str(y)) for (x,y) in peeps]
    
27. preds = [( x, str(y), 1) for (x,y) in preds]
    
28. 
    
29. 
    
30. def getKey(item):
    
31.   return item[1]
    
32. 
    
33. g = sorted(pop, key=getKey)
    
34. l = sorted(peeps, key=getKey)
    
35. pr = sorted(preds, key=getKey)
    
36. 
    
37. # add 3 points for our force graph centers
    
38. 
    
39. groups = ["0","1","2"] + [x[0] for x in g]
    
40. labels = ["0","1","2"] + [x[0] for x in l]
    
41. predictions = [[x[0],x[2]] for x in pr]
    
42. 
    
43. listIndices= list(range(3,len(predictions) + 3))
    
44. i = 0
    
45. for sublist in predictions:
    
46.   sublist.insert(0,listIndices[i])
    
47.   i += 1
    
48. >
    
49. 
    
50. %python
    
51. 
    
52. #create the viz
    
53. 
    
54. from lightning import Lightning
    
55. 
    
56. lgn = Lightning(host='http://public.lightning-viz.org')
    
57. lgn.create_session("new")
    
58. viz = lgn.force(predictions, group=groups, labels=labels)
    
59. viz.get_public_link()

复制代码

谢谢分享