我有一個看起來像這樣的資料框：
-------------- --------------------
|id | items |
-------------- --------------------
| 1|[a, b, .... x, y, z]|
-------------- --------------------
| 1|[q, z, .... x, b, 5]|
-------------- --------------------
| 2|[q, z, .... x, b, 5]|
-------------- --------------------
我想拆分行，使items列中的陣列的長度最多為 20。如果陣列的長度大于 20，我想創建新行并將陣列拆分，以便每個陣列的長度為 20 或更少. 因此，對于我的示例資料框中的第一行，如果我們假設長度為 10，并且我希望每行最多長度為 3，我希望它像這樣拆分：
-------------- --------------------
|id | items |
-------------- --------------------
| 1|[a, b, c] |
-------------- --------------------
| 1|[z, y, z] |
-------------- --------------------
| 1|[e, f, g] |
-------------- --------------------
| 1|[q] |
-------------- --------------------
理想情況下，如果陣列的長度不能被最大期望長度整除，則除最后一行之外的所有行的長度都應為 3。注意 - 該id列不是唯一的
uj5u.com熱心網友回復：
使用高階函式transform filter和slice，您可以將陣列拆分為大小為 20 的子陣列，然后將其分解：
val l = 20
val df1 = df.withColumn(
"items",
explode(
expr(
s"filter(transform(items, (x,i)-> IF(i%$l=0, slice(items,i 1,$l), null)), x-> x is not null)"
)
)
)
uj5u.com熱心網友回復：
你可以試試這個：
import pandas as pd
max_item_length = 3
df = pd.DataFrame(
{"fake_index": [1, 2, 3],
"items": [["a", "b", "c", "d", "e"], ["f", "g", "h", "i", "j"], ["k", "l"]]}
)
df2 = pd.DataFrame({"fake_index": [], "items": []})
for i in df.index:
try:
df2 = df2.append({"fake_index": int(df.iloc[i, 0]), "items": df.iloc[i, 1][:max_item_length]},
ignore_index=True)
df2 = df2.append({"fake_index": int(df.iloc[i, 0]), "items": df.iloc[i, 1][max_item_length:]},
ignore_index=True)
except:
df2 = df2.append({"fake_index": int(df.iloc[i, 0]), "items": df.iloc[i, 1]}, ignore_index=True)
df = df2
print(df)
輸入：
fake_index items
0 1 [a, b, c, d, e]
1 2 [f, g, h, i, j]
2 3 [k, l]
輸出：
fake_index items
0 1 [a, b, c]
1 1 [d, e]
2 2 [f, g, h]
3 2 [i, j]
4 3 [k, l]
uj5u.com熱心網友回復：
由于這需要更復雜的轉換，因此我使用了資料集。這可能沒有那么高效，但它會得到你想要的。
設定
創建一些示例資料來模擬您的資料。
val arrayData = Seq(
Row(1,List(1, 2, 3, 4, 5, 6, 7)),
Row(2,List(1, 2, 3, 4)),
Row(3,List(1, 2)),
Row(4,List(1, 2, 3))
)
val arraySchema = new StructType().add("id",IntegerType).add("values", ArrayType(IntegerType))
val df = spark.createDataFrame(spark.sparkContext.parallelize(arrayData), arraySchema)
/*
--- ---------------------
|id |values |
--- ---------------------
|1 |[1, 2, 3, 4, 5, 6, 7]|
|2 |[1, 2, 3, 4] |
|3 |[1, 2] |
|4 |[1, 2, 3] |
--- ---------------------
*/
轉型
// encoder for custom type of transformation
implicit val encoder = ExpressionEncoder[(Int, Array[Array[Int]])]
// Here we are using a sliding window of size 3 and step 3.
// This can be made into a generic function for a window of size k.
val df2 = df.map(r => {
val id = r.getInt(0)
val a = r.getSeq[Int](1).toArray
val arrays = a.sliding(3, 3).toArray
(id, arrays)
})
/*
--- ---------------------------------------------------------------
|_1 |_2 |
--- ---------------------------------------------------------------
|1 |[WrappedArray(1, 2, 3), WrappedArray(4, 5, 6), WrappedArray(7)]|
|2 |[WrappedArray(1, 2, 3), WrappedArray(4)] |
|3 |[WrappedArray(1, 2)] |
|4 |[WrappedArray(1, 2, 3)] |
--- ---------------------------------------------------------------
*/
val df3 = df2
.withColumnRenamed("_1", "id")
.withColumnRenamed("_2", "values")
/*
--- ---------------------------------------------------------------
|id |values |
--- ---------------------------------------------------------------
|1 |[WrappedArray(1, 2, 3), WrappedArray(4, 5, 6), WrappedArray(7)]|
|2 |[WrappedArray(1, 2, 3), WrappedArray(4)] |
|3 |[WrappedArray(1, 2)] |
|4 |[WrappedArray(1, 2, 3)] |
--- ---------------------------------------------------------------
*/
使用爆炸
Expode將為第二列中的每個陣列條目創建一個新元素。
val df4 = df3.withColumn("values", functions.explode($"values"))
/*
--- ---------
|id |values |
--- ---------
|1 |[1, 2, 3]|
|1 |[4, 5, 6]|
|1 |[7] |
|2 |[1, 2, 3]|
|2 |[4] |
|3 |[1, 2] |
|4 |[1, 2, 3]|
--- ---------
*/
限制
這種方法并非沒有限制。
Primarily, it will not be as performant on larger datasets since this code is no longer using dataframe built-in optimizations. However, the dataframe API might require the use of window functions, which can also have limited performance based on the size of the data. If it's possible to alter this data at the source, this would be recommended.
This approach also requires defining an encoder for something more complex. If the data schema changes, then different encoders will have to be used.
轉載請註明出處，本文鏈接：https://www.uj5u.com/gongcheng/424397.html
標籤：数据框 斯卡拉 阿帕奇火花 apache-spark-sql
上一篇：根據值減少迭代器