代码之家 › 专栏 › 技术社区 › Justin Pihony

使用地图创建新列

apache-spark-sql dataframe apache-spark

Justin Pihony · 技术社区 · 9 年前

是否有一种方法(不使用UDF)来获取现有数据帧并通过获取现有列并从映射中提取其等效值来创建新列?

df.withColumn("newCol", transform(col("existing").using(map)))

哪里 map 的密钥类型与 existing ,值是我想要的输出。

2 回复 | 直到 9 年前

zero323 little_kid_pea 9 年前

您可以转换 Map 到 DataFrame 和 join :

val df = sc.parallelize(Seq(
    (1, "foo"), (2, "bar"), (3, "foobar")
)).toDF("id", "existing")

val map = Map("foo" -> 1, "bar" -> 2)
val lookup = sc.parallelize(map.toSeq).toDF("key", "value")

df
 .join(lookup, $"existing" <=> $"key", "left")
 .drop("key")
 .withColumnRenamed("value", "newCol")

zero323 little_kid_pea 9 年前

import sqtx.implicits._
val x = Map("foo" -> 1,"bar"-> 2, "baz"->3)

val df = sc.parallelize(Seq(
  (1, "foo"), (2, "bar"), (3, "foobar")
)).toDF("id", "existing")
df.map(r => (r.getInt(0),x.getOrElse(r.getString(1),0))).toDF("id","new")

推荐文章

srinath tripuraneni · {DataFrameWriter CSV到HDFS文件系统}不分区写入数据

3 年前

Calcutta · Google Colab中的Spark SQL在大数据上失败

3 年前

Palkin Jangra · 使用循环在Pyspark数组元素上和元素本身内迭代两次

3 年前

Doraemon · PySpark:使用不同值的字符串类型列创建聚合列

3 年前

amol visave · spark作业失败时会发生什么?

3 年前

Alex Jolly · 如何在另一个pyspark数据帧中查询开始时间和结束时间之间的日期时间

3 年前

chun · pyspark dataframe在s3中两次写入csv文件

3 年前

Mod_x · 如何将特定列的行标题更改为行标题,并在pySpark中生成矩阵?

3 年前

katty · 动态地将参数传递给scala中的函数

6 年前

Abhishek Choudhary · 如何更新数组列?

6 年前