如何将地图转换为 Spark 的 RDD

Posted

技术标签:

【中文标题】如何将地图转换为 Spark 的 RDD【英文标题】:How to convert a map to Spark's RDD 【发布时间】:2015-11-11 21:06:59 【问题描述】:

我有一个嵌套映射形式的数据集,它的 Scala 类型是:

Map[String, (LabelType,Map[Int, Double])]

第一个String键是每个样本的唯一标识符,值是一个包含标签(为-1或1)的元组,以及一个嵌套映射,它是非零的稀疏表示与样本相关的元素。

我想将这些数据加载到 Spark(使用 MUtil)并训练和测试一些机器学习算法。

使用 LibSVM 的稀疏编码很容易将这些数据写入文件,然后在 Spark 中加载:

writeMapToLibSVMFile(data_map,"libsvm_data.txt") // Implemeneted some where else
val conf = new SparkConf().setAppName("DecisionTree").setMaster("local[4]")
val sc = new SparkContext(conf)

// Load and parse the data file.
val data = MLUtils.loadLibSVMFile(sc, "libsvm_data.txt")
// Split the data into training and test sets
val splits = data.randomSplit(Array(0.7, 0.3))
val (trainingData, testData) = (splits(0), splits(1))

// Train a DecisionTree model.

我知道直接从data_map 加载data 变量应该很容易,但我不知道怎么做。

感谢任何帮助!

【问题讨论】:

【参考方案1】:

可以通过两种方式完成

    sc.textFile("libsvm_data.txt").map(s => createObject()) 将地图转换为对象集合并使用sc.parallelize()

首选第一个。

【讨论】:

请注意“libsvm_data.txt”需要先写入文件,我想避免。 没错,必须避免。【参考方案2】:

我猜你想要这样的东西

import org.apache.spark.rdd.RDD
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.regression.LabeledPoint

// If you know this upfront, otherwise it can be computed
// using flatMap
// trainMap.values.flatMap(_._2.keys).max + 1
val nFeatures: Int = ??? 

val trainMap = Map(
  "x001" -> (-1, Map(0 -> 1.0, 3 -> 5.0)),
  "x002" -> (1, Map(2 -> 5.0, 3 -> 6.0)))

val trainRdd: RDD[(String, LabeledPoint)]  = sc
  // Convert Map to Seq so it can passed to parallelize
  .parallelize(trainMap.toSeq)
  .mapcase (id, (labelInt, values)) => 

      // Convert nested map to Seq so it can be passed to Vector
      val features = Vectors.sparse(nFeatures, values.toSeq)

      // Convert label to Double so it can be used for LabeledPoint
      val label = labelInt.toDouble 

      (id, LabeledPoint(label, features))
 

【讨论】:

以上是关于如何将地图转换为 Spark 的 RDD的主要内容,如果未能解决你的问题,请参考以下文章

如何解决嵌套地图函数中的 SPARK-5063

Spark - 地图转换

哪个选项使用 pyspark 提供最佳性能?使用地图进行 UDF 或 RDD 处理?

Scala - 将 RDD[String] 转换为地图

Scala Spark 地图类型匹配问题

如何将地图 List<Map<String, String>> myList 列表转换为 Java 中的 Spark Dataframe?