Блог

У меня возникли проблемы, связанные с производительностью, при объединении двойного массива путем поиска в массиве индексов. Что я имею в виду под этим, так это. исходный кадр данных выглядит примерно так:

 original Dataframe 

| id | prop1        | values                  |
|----|--------------|-------------------------|
|  1 | [2,5,1,3]    |   [ 0.1, 0.5, 0.7, 0.8] |
|  2 | [2,1]        |   [ 0.2, 0.3 ]          |
|  1 | [1,5]        |   [ 0.4, 0.3 ]          |
|  2 | [3,2]        |   [ 0.0, 0.1 ]          |

so in the column 2 which is prop1 is an int array having values within range of 1 to 5 but not in a order and there can be missing numbers within array.
 

Массив Prop1 int похож на индекс значений двойного массива
под этим я подразумеваю, что строка 1 выглядит примерно так, как показано ниже, когда она разнесена

 | id | prop1 | values |
|----|-------|--------|
|  1 | 2     |   0.1  |
|  1 | 5     |   0.5  |
|  1 | 1     |   0.7  |
|  1 | 3     |   0.8  |
 

Последняя проблема,

поэтому мне нужно агрегировать значения двойного массива, заглянув в массив индексов и по идентификатору столбца

таким образом, результат должен быть

 | id | prop1          | values                   | 
|----|----------------|--------------------------| 
|  1 | [2,5,1,3]      |   [ 0.1, 0.8, 1.1, 0.8 ] | 
|  2 | [2,1,3]        |   [ 0.3, 0.3, 0.0 ]      | 


Below code I am using to extract the values by index and pivot right before merging them to array

//dummy dataframe to get the sequence of 5 but the upper end is dynamic value and that can extend till 300k
var df = (1 to 5).toDF("prop1")

//joining original Df by prop1 column 
var stgDf = originalDf.join(df,originalDf.col("prop1") ===  df.col("prop1"),"inner")

// pivoting the values by index
var pivotDf = stgDf.groupBy("id")
             .pivot("prop1").agg(first("values"))

 // now aggregating the pivoted  values by id
 var expr = pivtoDf.columns.map(sum(_))
 var pivotDf.groupBy("id").agg(expr.head,expr.tail:_*)

 //then grouping back into array by id
 

Это решение я сделал, используя взрыв prop1 и значение, оно работало с несколькими строками, но в реальной задаче массивы обоих столбцов могут превышать 500 тысяч значений каждый, а количество строк на идентификатор может превышать 30 миллионов

Если кто-нибудь может посмотреть и помочь в этом, это было бы здорово. приложение построено в scala с использованием spark 2.4

заранее спасибо

Для версии v3.x, а не для версии v2.4. Обновление, как слишком сложное.

Какие-то серьезные споры о данных!

Вероятно, есть лучшие способы, но они масштабируемы. Может потребоваться много разделов.

 import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._
import org.apache.spark.sql.Column

val arrayStructureData = Seq(
Row(1,List(2,5,1,3),List(0.1, 0.5, 0.7, 0.8)),
Row(2,List(2,1),List(0.2, 0.3)),
Row(1,List(1,5),List(0.4, 0.3)),
Row(2,List(3,2),List(0.0, 0.1)) 
)
// Just a single StructType for the Row
val arrayStructureSchema = new StructType()
    .add("id",IntegerType)
    .add("prop1", ArrayType(IntegerType))
    .add("values", ArrayType(DoubleType))
val df = spark.createDataFrame(spark.sparkContext.parallelize(arrayStructureData),arrayStructureSchema)
df.printSchema()
df.show()

val df2 = df.withColumn(
  "jCols",
  zip_with(
    col("prop1"),
    col("values"),
      // Should be a struct really, but...array used. zip_with not available in v2.4!
      (left: Column, right: Column) => array(left, right)
  )
).drop('prop1).drop('values)

df2.show(false)
df2.printSchema()

val df3 = df2.groupBy("id").agg(collect_list("jCols").as("jCols"))
df3.printSchema()
df3.show(false)

val df4 = df3.select($"id",flatten($"jCols").as("jCols"))
df4.show(false)
df4.printSchema()

val df5 = df4.withColumn("ExjCols", explode($"jCols")).drop("jCols")
df5.show(false)
df5.printSchema()

val df6 = df5.select(col("id"),col("ExjCols")(0).as("prop1"),col("ExjCols")(1).as("values"))
df6.show(false)
df6.printSchema()

val df7 = df6.groupBy("id", "prop1").sum("values").toDF("id","prop1","values") 
df7.show(false)
df7.printSchema()

val df8 = df7.withColumn("combined", array($"prop1", $"values"))
df8.show(false)
df8.printSchema()

val df9 = df8.groupBy("id").agg(collect_list("combined").as("propN"))
df9.show(false)
df9.printSchema()

val res = df9.withColumn("prop1",expr("transform(propN, x -> x[0])")).withColumn("values",expr("transform(propN, x -> x[1])")).drop('propN)
res.show(false)
 

ВОЗВРАТ:

  --- -------------------- ------------------------------- 
|id |prop1               |values                         |
 --- -------------------- ------------------------------- 
|1  |[2.0, 5.0, 1.0, 3.0]|[0.1, 0.8, 1.1, 0.8]           |
|2  |[2.0, 1.0, 3.0]     |[0.30000000000000004, 0.3, 0.0]|
 --- -------------------- ------------------------------- 
 

Понятия не имею, почему точность 0,3000… произошло, но это произошло. Также исправил пример, в нем были некоторые ошибки.

Я могу только предположить, что сейчас это менее популярно, так как для ответа потребовалось некоторое время.