Spark collect_list เปลี่ยน data_type จากอาร์เรย์เป็นสตริง

Question

ฉันมีการรวมดังต่อไปนี้

val df_date_agg = df
    .groupBy($"a",$"b",$"c") .agg(sum($"d").alias("data1"),sum($"e").alias("data2")) .groupBy($"a")
    .agg(collect_list(array($"b",$"c",$"data1")).alias("final_data1"), collect_list(array($"b",$"c",$"data2")).alias("final_data2"))

ที่นี่ฉันกำลังทำการรวบรวมและรวบรวมผลลัพธ์ด้วยcollect_list. ก่อนหน้านี้เราใช้ spark 1 และให้ประเภทข้อมูลด้านล่าง

 |-- final_data1: array (nullable = true)
 |    |-- element: string (containsNull = true)
 |-- final_data2: array (nullable = true)
 |    |-- element: string (containsNull = true)

ตอนนี้เราต้องย้ายไปที่จุดประกาย 2 แต่เราได้รับสคีมาด้านล่าง

|-- final_data1: array (nullable = true)
 |    |-- element: array (containsNull = true)
 |    |    |-- element: string (containsNull = true)
 |-- final_data1: array (nullable = true)
 |    |-- element: array (containsNull = true)
 |    |    |-- element: string (containsNull = true)

ในการfirst()บันทึกด้านล่างคือความแตกต่าง

spark 1.6

[2020-09-26, Ayush, 103.67] => datatype string

spark 2 

WrappedArray(2020-09-26, Ayush, 103.67)

ฉันจะเก็บข้อมูลประเภทเดียวกันได้อย่างไร

แก้ไข - พยายามใช้ Concat

วิธีหนึ่งที่ฉันได้สคีมาที่แน่นอนเช่น Spark 1.6 คือการใช้ concat แบบนี้

val df_date_agg = df
    .groupBy($"msisdn",$"event_date",$"network") .agg(sum($"data_mou").alias("data_mou_dly"),sum($"voice_mou").alias("voice_mou_dly")) .groupBy($"msisdn")
    .agg(collect_list(concat(lit("["),lit($"event_date"),lit(","),lit($"network"),lit(","),lit($"data_mou_dly"),lit("]")))

จะส่งผลต่อประสิทธิภาพโค้ดของฉันหรือไม่? มีวิธีที่ดีกว่านี้หรือไม่?

Answer 1

เนื่องจากคุณต้องการเป็นตัวแทนสตริงของอาร์เรย์แล้วการแคสต์อาร์เรย์เป็นสตริงแบบนี้ล่ะ

val df_date_agg = df
    .groupBy($"a",$"b",$"c") .agg(sum($"d").alias("data1"),sum($"e").alias("data2")) .groupBy($"a")
    .agg(collect_list(array($"b",$"c",$"data1") cast "string").alias("final_data1"), collect_list(array($"b",$"c",$"data2") cast "string").alias("final_data2"))

อาจเป็นเพียงสิ่งที่ Spark เวอร์ชันเก่าของคุณกำลังทำอยู่

วิธีแก้ปัญหาที่คุณเสนออาจใช้งานได้ดี แต่litไม่จำเป็นต้องตัดการอ้างอิงคอลัมน์ของคุณ( lit($"event_date")) $"event_date"ก็เพียงพอแล้ว

Answer 2

Fllttening final1 และ final2 คอลัมน์จะแก้ไขปัญหานี้

val data = Seq((1,"A", "B"), (1, "C", "D"), (2,"E", "F"), (2,"G", "H"), (2,"I", "J"))

val df = spark.createDataFrame(
  data
).toDF("col1", "col2", "col3")

val old_df = df.groupBy(col("col1")).agg(
    collect_list(
        array(
            col("col2"), 
            col("col3")
            )
    ).as("final")
    )
val new_df = old_df.select(col("col1"), flatten(col("final")).as("final_new"))
println("Input Dataframe")

df.show(false)
println("Old schema format")
old_df.show(false)
old_df.printSchema()

println("New schema format")
new_df.show(false)
new_df.printSchema()

เอาท์พุต:

Input Dataframe
+----+----+----+
|col1|col2|col3|
+----+----+----+
|1   |A   |B   |
|1   |C   |D   |
|2   |E   |F   |
|2   |G   |H   |
|2   |I   |J   |
+----+----+----+

Old schema format
+----+------------------------+
|col1|final                   |
+----+------------------------+
|1   |[[A, B], [C, D]]        |
|2   |[[E, F], [G, H], [I, J]]|
+----+------------------------+

root
 |-- col1: integer (nullable = false)
 |-- final: array (nullable = true)
 |    |-- element: array (containsNull = true)
 |    |    |-- element: string (containsNull = true)

New schema format
+----+------------------+
|col1|final_new         |
+----+------------------+
|1   |[A, B, C, D]      |
|2   |[E, F, G, H, I, J]|
+----+------------------+

root
 |-- col1: integer (nullable = false)
 |-- final_new: array (nullable = true)
 |    |-- element: string (containsNull = true)

ในกรณีของคุณ specefic

val df_date_agg = df
    .groupBy($"a",$"b",$"c") .agg(sum($"d").alias("data1"),sum($"e").alias("data2")) .groupBy($"a")
    .agg(collect_list(array($"b",$"c",$"data1")).alias("final_data1"), collect_list(array($"b",$"c",$"data2")).alias("final_data2"))
         .select(flatten(col("final_data1").as("final_data1"), flatten(col("final_data2).as("final_data2))