如何将行 ID 的持久列添加到 Spark 数据帧

这个问题并不新鲜，但是我在Spark中发现了令人惊讶的行为。我需要向数据帧添加一列行 ID。我使用了 DataFrame 方法 monotonically_increasing_id（），它确实给了我一个额外的唯一行 ID 列（顺便说一下，它们不是连续的，但是唯一的）。

我遇到的问题是，当我过滤数据帧时，生成的数据帧中的行 ID 被重新分配。两个数据帧如下所示。

第一个是初始数据帧，其中添加了行 ID，如下所示：
```
df.withColumn("rowId", monotonically_increasing_id()) 
```
第二个数据帧是通过 df.filter(col("P")) 对 col P 进行过滤后获得的数据帧。

custId 169 的 rowId 说明了这个问题，它在初始数据帧中为 5，但在过滤后，当 custId 169 被过滤掉时，rowId （5）被重新分配给 custmId 773！我不知道为什么这是默认行为。

我希望rowIds是"粘性的";如果我从数据帧中删除行，我不希望它们的 ID "重复使用"，我希望它们也随行一起消失。有可能做到吗？我没有看到任何标志可以从monotonically_increasing_id方法请求此行为。

+---------+--------------------+-------+
| custId  |    features|    P  |rowId|
+---------+--------------------+-------+
|806      |[50,5074,...|   true|    0|
|832      |[45,120,1...|   true|    1|
|216      |[6691,272...|   true|    2|
|926      |[120,1788...|   true|    3|
|875      |[54,120,1...|   true|    4|
|169      |[19406,21...|  false|    5|
after filtering on P:
+---------+--------------------+-------+
|   custId|    features|    P  |rowId|
+---------+--------------------+-------+
|      806|[50,5074,...|   true|    0|
|      832|[45,120,1...|   true|    1|
|      216|[6691,272...|   true|    2|
|      926|[120,1788...|   true|    3|
|      875|[54,120,1...|   true|    4|
|      773|[3136,317...|   true|    5|

Spark 2.0

此问题已在 Spark 2.0 中与 SPARK-14241 一起解决。
另一个类似的问题已在Spark 2.1中使用SPARK-14393得到解决

火花 1.x

你遇到的问题相当微妙，但可以简化为一个简单的事实monotonically_increasing_id这是一个极其丑陋的功能。它显然不是纯粹的，它的价值取决于你完全无法控制的东西。

它

不需要任何参数，因此从优化器的角度来看，何时调用它并不重要，并且可以在所有其他操作之后推送。因此，您看到的行为。

如果您查看代码，您会发现这是通过使用 Nondeterministic 扩展表达式来显式标记MonotonicallyIncreasingID的。

~~我认为没有任何优雅的解决方案，但您可以处理此问题的一种方法是添加对过滤值的人为依赖。例如，对于这样的 UDF：~~

from pyspark.sql.types import LongType
from pyspark.sql.functions import udf
bound = udf(lambda _, v: v, LongType()) 
(df
  .withColumn("rn", monotonically_increasing_id())
  # Due to nondeterministic behavior it has to be a separate step
  .withColumn("rn", bound("P", "rn"))  
  .where("P"))

一般来说，在RDD上使用zipWithIndex添加索引，然后将其转换回DataFrame可能会更干净。

* 上面显示的解决方法在 Spark 2.x 中不再是有效的解决方案（也不是必需的），其中 Python UDF 是执行计划优化的主题。

我无法重现这一点。我使用的是Spark 2.0，所以可能行为已经改变，或者我没有和你做同样的事情。

val df = Seq(("one", 1,true),("two", 2,false),("three", 3,true),("four", 4,true))
.toDF("name", "value","flag")
.withColumn("rowd", monotonically_increasing_id())
df.show
val df2 = df.filter(col("flag")=== true)
df2.show
df: org.apache.spark.sql.DataFrame = [name: string, value: int ... 2 more fields]
+-----+-----+-----+----+
| name|value| flag|rowd|
+-----+-----+-----+----+
|  one|    1| true|   0|
|  two|    2|false|   1|
|three|    3| true|   2|
| four|    4| true|   3|
+-----+-----+-----+----+
df2: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [name: string, value: int ... 2 more fields]
+-----+-----+----+----+
| name|value|flag|rowd|
+-----+-----+----+----+
|  one|    1|true|   0|
|three|    3|true|   2|
| four|    4|true|   3|
+-----+-----+----+----+

我最近在研究一个类似的问题。虽然monotonically_increasing_id()非常快，但它并不可靠，不会给你连续的行号，只会增加唯一的整数。

创建 Windows 分区然后使用row_number().over(some_windows_partition)非常耗时。

到目前为止，最好的解决方案是使用带有索引的压缩，然后将压缩文件转换回原始数据帧，新架构包括索引列。

试试这个：

from pyspark.sql import Row
from pyspark.sql.types import StructType, StructField, LongType
new_schema = StructType(**original_dataframe**.schema.fields[:] + [StructField("index", LongType(), False)])
zipped_rdd = **original_dataframe**.rdd.zipWithIndex()
indexed = (zipped_rdd.map(lambda ri: row_with_index(*list(ri[0]) + [ri[1]])).toDF(new_schema))

其中original_dataframe是您必须添加索引dataframe，row_with_index是具有列索引的新架构，您可以将其写入为

row_with_index = Row(
"calendar_date"
,"year_week_number"
,"year_period_number"
,"realization"
,"index"
)

在这里，calendar_date、year_week_number、year_period_number和realization是我原来dataframe的专栏。您可以将名称替换为列的名称。索引是必须为行号添加的新列名。

与row_number().over(some_windows_partition)方法相比，此过程在很大程度上更有效，更顺畅。

希望这有帮助。

要绕过 monotonically_increasing_id（）的移位计算，您可以尝试将数据帧写入磁盘并重新读取。然后，id 列现在只是一个正在读取的数据字段，而不是在管道中的某个点动态计算。虽然这是一个非常丑陋的解决方案，但当我进行快速测试时它有效。

这对

我有用。创建了另一个标识列并使用了窗口函数row_number

import org.apache.spark.sql.functions.{row_number}
import org.apache.spark.sql.expressions.Window
val df1: DataFrame = df.withColumn("Id",lit(1))
df1
.select(
...,
row_number()
.over(Window
.partitionBy("Id"
.orderBy(col("...").desc))
)
.alias("Row_Nbr")
)

为了获得更好的性能 wrt Chris T 解决方案，您可以尝试写入 apache ignite 共享数据帧而不是写入磁盘。https://ignite.apache.org/use-cases/spark/shared-memory-layer.html

最好的方法是使用唯一键的 concat 哈希。

例如：在python中：

from pyspark.sql.functions import concat, md5
unique_keys = ['event_datetime', 'ingesttime']
raw_df.withColumn('rowid', md5(concat(*unique_keys)))

原因如下：

新的"rowid"是从输入数据中确定性派生的（相对于uuid，这是不确定的）
附加新数据很容易。（与其他方式相比：如monotonically_increasing_id（）或row_number（），需要获取当前的最大数字..
仅供参考 https://bzhangusc.wordpress.com/2016/03/23/create-unique-record-key-for-table-linking/

相关内容

最新更新

热门标签：