下面的問題有 scala 和 pyspark 的解決方案，這個問題中提供的解決方案不適用于連續的索引值。Spark Dataframe：如何添加索引列：又名分布式數據索引我在 Apache-spark 中有一個現有數據集，我想根據索引從中選擇一些行。我打算添加一個索引列，其中包含從 1 開始的唯一值，并根據該列的值獲取行。我發現以下方法可以添加使用排序依據的索引：df.withColumn("index", functions.row_number().over(Window.orderBy("a column")));我不想使用排序依據。我需要索引的順序與它們在數據集中的順序相同。有什么幫助嗎？
2 回答
阿晨1998
TA貢獻2037條經驗 獲得超6個贊
據我所知，您正在嘗試將索引（具有連續值）添加到數據框。不幸的是，在 Spark 中沒有內置函數可以做到這一點。您只能使用 df.withColumn("index", ) 添加遞增索引（但不一定具有連續值monotonicallyIncreasingId）。
盡管如此，RDD API 中有一個zipWithIndex函數可以完全滿足您的需要。因此，我們可以定義一個函數，將數據幀轉換為 RDD，添加索引并將其轉換回數據幀。
我不是 java 中 spark 的專家（scala 更緊湊）所以可能會做得更好。這是我會怎么做。
public static Dataset<Row> zipWithIndex(Dataset<Row> df, String name) {
JavaRDD<Row> rdd = df.javaRDD().zipWithIndex().map(t -> {
Row r = t._1;
Long index = t._2 + 1;
ArrayList<Object> list = new ArrayList<>();
r.toSeq().iterator().foreach(x -> list.add(x));
list.add(index);
return RowFactory.create(list);
});
StructType newSchema = df.schema()
.add(new StructField(name, DataTypes.LongType, true, null));
return df.sparkSession().createDataFrame(rdd, newSchema);
}
以下是您將如何使用它。請注意內置 spark 函數的作用與我們的方法的作用形成對比。
Dataset<Row> df = spark.range(5)
.withColumn("index1", functions.monotonicallyIncreasingId());
Dataset<Row> result = zipWithIndex(df, "good_index");
// df
+---+-----------+
| id| index1|
+---+-----------+
| 0| 0|
| 1| 8589934592|
| 2|17179869184|
| 3|25769803776|
| 4|25769803777|
+---+-----------+
// result
+---+-----------+----------+
| id| index1|good_index|
+---+-----------+----------+
| 0| 0| 1|
| 1| 8589934592| 2|
| 2|17179869184| 3|
| 3|25769803776| 4|
| 4|25769803777| 5|
+---+-----------+----------+
UYOU
TA貢獻1878條經驗 獲得超4個贊
上面的答案經過一些調整對我有用。下面是一個功能性的 Intellij Scratch 文件。我在 Spark 2.3.0 上：
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.functions;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.Metadata;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import java.util.ArrayList;
class Scratch {
public static void main(String[] args) {
SparkSession spark = SparkSession
.builder()
.appName("_LOCAL")
.master("local")
.getOrCreate();
Dataset<Row> df = spark.range(5)
.withColumn("index1", functions.monotonicallyIncreasingId());
Dataset<Row> result = zipWithIndex(df, "good_index");
result.show();
}
public static Dataset<Row> zipWithIndex(Dataset<Row> df, String name) {
JavaRDD<Row> rdd = df.javaRDD().zipWithIndex().map(t -> {
Row r = t._1;
Long index = t._2 + 1;
ArrayList<Object> list = new ArrayList<>();
scala.collection.Iterator<Object> iterator = r.toSeq().iterator();
while(iterator.hasNext()) {
Object value = iterator.next();
assert value != null;
list.add(value);
}
list.add(index);
return RowFactory.create(list.toArray());
});
StructType newSchema = df.schema()
.add(new StructField(name, DataTypes.LongType, true, Metadata.empty()));
return df.sparkSession().createDataFrame(rdd, newSchema);
}
}
輸出：
+---+------+----------+
| id|index1|good_index|
+---+------+----------+
| 0| 0| 1|
| 1| 1| 2|
| 2| 2| 3|
| 3| 3| 4|
| 4| 4| 5|
+---+------+----------+
添加回答
舉報
0/150
提交
取消