I have a requirement to add sequence number to the columns (Array) where my source data is in parquet format and volume is nearly 2 Billion records. where i have to choose only key and code column from parquet and add sequence number to ref_codes and load it to back to S3
Key_1 Key_2 Key_3 Ref_codes
112240386 7435038894 2 [4659,53540,78907]
113325994 7940375640 1 [7232,7840,83969]
223352476 7765270324 4 [9999]
345936074 7950076012 1 [78650,4829,30000]
Key_1 Key_2 Key_3 Ref_codes
112240386 7435038894 2 [(4659,0),(53540,1),(78907,2)]
113325994 7940375640 1 [(7232,0),(7840,1),(83969,2)]
223352476 7765270324 4 [(9999,0)]
345936074 7950076012 1 [(78650,0),(4829,1),(30000,2)]
I am new to Scala, I have tried multiple option but doesn't got the proper results.Any help really appreciated...
You can use higher-order functions like transform in the latest versions of spark as below.
Data:
val df = Seq(
("112240386", "7435038894", 2, Array(4659, 53540, 7890)),
("113325994", "7940375640", 1, Array(7232, 7840, 8396)),
("223352476", "7765270324", 4, Array(999)),
("345936074", "7950076012", 1, Array(78650, 4829, 3000)),
).toDF("key_1", "key_2", "key_3", "ref_code")
Spark 3.0.0+
df.withColumn("ref_code", transform($"ref_code", (x, i) => struct(x, i) ))
Spark > 2.4
df.withColumn("ref_code", expr("transform(ref_code, (x,i) -> (x,i) )"))
Spark < 2.4
val addIndex = udf((arr: Seq[Int]) => arr.zipWithIndex)
df.withColumn("ref_code", addIndex($"ref_code")).show(false)
Output:
+---------+----------+-----+----------------------------------+
|key_1 |key_2 |key_3|ref_code |
+---------+----------+-----+----------------------------------+
|112240386|7435038894|2 |[[4659, 0], [53540, 1], [7890, 2]]|
|113325994|7940375640|1 |[[7232, 0], [7840, 1], [8396, 2]] |
|223352476|7765270324|4 |[[999, 0]] |
|345936074|7950076012|1 |[[78650, 0], [4829, 1], [3000, 2]]|
+---------+----------+-----+----------------------------------+
More example on transform is here