下面是在pyspark中执行相同操作的说明。我同意拉姆的解释。 collectAsMap仅适用于pairedrdd,因此您需要先将数据帧转换为pairedrdd,然后使用collectAsMap函数将其转换为一些字典。

例如,我有一个下面的数据框:

df = spark.sql("""select emp_id,emp_city from udb.temptable_1 order by emp_id""");
+------+--------+
|emp_id|emp_city|
+------+--------+
|     1|NOIDA   |
|     2|GURGAON |
|     3|DWARKA  |
|     4|SAKET   |
|     5|USA     |
|     6|UP      |
|     7|NOIDA   |
|     8|SAKET   |
|     9|GURGAON |
+------+--------+

将其转换为键值对rdd

newrdd = df.rdd.map(lambda x : (x[0],x))

>>> type(newrdd)
<class 'pyspark.rdd.PipelinedRDD'>

[(1, Row(emp_id=1, emp_city=u'NOIDA   ')), 
(2, Row(emp_id=2, emp_city=u'GURGAON ')), 
(3, Row(emp_id=3, emp_city=u'DWARKA  ')), 
(4, Row(emp_id=4, emp_city=u'SAKET   ')), 
(5, Row(emp_id=5, emp_city=u'USA     ')), 
(6, Row(emp_id=6, emp_city=u'UP      ')), 
(7, Row(emp_id=7, emp_city=u'NOIDA   ')), 
(8, Row(emp_id=8, emp_city=u'SAKET   ')), 
(9, Row(emp_id=9, emp_city=u'GURGAON '))]

最后,您可以使用collectAsMap将键值对rdd转换为dict

dict = newrdd.collectAsMap()

{1: Row(emp_id=1, emp_city=u'NOIDA   '), 
2: Row(emp_id=2, emp_city=u'GURGAON '), 
3: Row(emp_id=3, emp_city=u'DWARKA  '), 
4: Row(emp_id=4, emp_city=u'SAKET   '), 
5: Row(emp_id=5, emp_city=u'USA     '), 
6: Row(emp_id=6, emp_city=u'UP      '), 
7: Row(emp_id=7, emp_city=u'NOIDA   '), 
8: Row(emp_id=8, emp_city=u'SAKET   '), 
9: Row(emp_id=9, emp_city=u'GURGAON ')}

>>> dict.keys()
[1, 2, 3, 4, 5, 6, 7, 8, 9]

>>> dict.get(2)
Row(emp_id=2, emp_city=u'GURGAON ')

首先,我在python/pyspark方面不好,所以我演示了使用scala。。。

collectAsMap 仅适用于pairedRDD(参见下面的代码 火花 doc /代码库)

/**
       * Return the key-value pairs in this RDD to the master as a Map.
       *
       * Warning: this doesn't return a multimap (so if you have multiple values to the same key, only
       *          one value per key is preserved in the map returned)
       *
       * @note this method should only be used if the resulting data is expected to be small, as
       * all the data is loaded into the driver's memory.
       */
      def collectAsMap(): Map[K, V] = self.withScope {
        val data = self.collect()
        val map = new mutable.HashMap[K, V]
        map.sizeHint(data.length)
        data.foreach { pair => map.put(pair._1, pair._2) }
        map
      }

你的 df.rdd 相当于 RDD[Row] 既然你要皈依 DataFrame 到 RDD .

所以你不能把它收集成地图。除非你必须做一个 keyBy ( 通过应用 f )该行中的任何元素都将指向该行。或者其他一些将其转换为成对RDD的操作。

下面是一个完整的例子来证明这一点。

import org.apache.log4j.{Level, Logger}
import org.apache.spark.internal.Logging
import org.apache.spark.sql.SparkSession

/** *
  * collectAsMap is only applicable to pairedrdd if you want to do a map then you can do a rdd key by and proceed
  *
  * @author : Ram Ghadiyaram
  */
object PairedRDDPlay extends Logging {
  Logger.getLogger("org").setLevel(Level.OFF)
  // Logger.getLogger("akka").setLevel(Level.OFF)

  def main(args: Array[String]): Unit = {
    val appName = if (args.length > 0) args(0) else this.getClass.getName
    val spark: SparkSession = SparkSession.builder
      .config("spark.master", "local") //.config("spark.eventLog.enabled", "true")
      .appName(appName)
      .getOrCreate()
    import spark.implicits._
    val pairs = spark.sparkContext.parallelize(Array((1, 1,3), (1, 2,3), (1, 3,3), (1, 1,3), (2, 1,3))).toDF("mycol1", "mycol2","mycol3")
    pairs.show()

    val keyedBy = pairs.rdd.keyBy(_.getAs[Int]("mycol1"))
    keyedBy.foreach(x => println("using keyBy-->>" + x))
    val myMap = keyedBy.collectAsMap()
    println(myMap.toString())
    assert(myMap.size == 2)
    //    val myMap1 = pairs.rdd.collectAsMap()
    //    println(myMap1.toString())
    //    assert(myMap1.size == 2)
    //Error:(28, 28) value collectAsMap is not a member of org.apache.spark.rdd.RDD[org.apache.spark.sql.Row]
    //    val myMap1 = pairs.rdd.collectAsMap()
  }
}

结果:

Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
+------+------+------+
|mycol1|mycol2|mycol3|
+------+------+------+
|     1|     1|     3|
|     1|     2|     3|
|     1|     3|     3|
|     1|     1|     3|
|     2|     1|     3|
+------+------+------+

using keyBy-->>(1,[1,1,3])
using keyBy-->>(1,[1,2,3])
using keyBy-->>(1,[1,3,3])
using keyBy-->>(1,[1,1,3])
using keyBy-->>(2,[2,1,3])
Map(2 -> [2,1,3], 1 -> [1,1,3])

问题:DF有8个字段,这是否意味着 收集地图 ()可以使用 只有两个字段?

答:不,您可以在示例中看到具有多列(即>2)的示例。但你需要把它转换成pairrdd。

还可以看看 how-does-the-collectasmap-function-work-for-spark-api