I have some Kafka Topics I need to consume using spark scala. Within the schema of the topic, there are multiple Arrays that represents entities that need to be broken out into separate dataframes and eventually into separate tables. The code I have works but not recursively. Nested arrays within arrays are not working as expected.
The first thing the code does is takes the raw schema and flattens out any attributes within a given Struct that isn't an array (function defined below: flattenSchema). Then the code tries to determine how many arrays exist within all of the original dataframe (function below: getArraysFromDataframe) so it can iterate on them and then generate new dataframes that are exploded for writing to new tables. When the code reaches nested Arrays it fails on the explode. I think because the flattenSchema function isn't flattening out attributes in the nested arrays multiple levels down.
I found the need to use the getArraysFromDataframe function because doing a .dtypes.foreach on the flattened dataframe did not work because the arrays that were nested inside other arrays were not iterated on. So child arrays were ignored because they are apart of the parent array and I needed to recursively go through each level to find the child arrays.
Here's the code:
Functions
def flattenSchema(schema: StructType, prefix: String = null) : Array[Column] = {
schema.fields.flatMap(f => {
val colName = if (prefix == null) f.name else (prefix + "." + f.name)
f.dataType match {
case st: StructType => flattenSchema(st, colName)
case _ => Array(col(colName))
}
})
}
def getArraysFromDataframe(df: DataFrame, lb: ListBuffer[String]): ListBuffer[String] = {
//getting all the fields from schema
val fields = df.schema.fields
val fieldNames = fields.map(x => x.name)
//length shows the number of fields inside dataframe
val length = fields.length
for (i <- 0 to fields.length - 1) {
val field = fields(i)
val fieldtype = field.dataType
val fieldName = field.name
fieldtype match {
case arrayType: ArrayType =>
val fieldName1 = fieldName
val fieldNamesExcludingArray = fieldNames.filter(_ != fieldName1)
val fieldNamesAndExplode = fieldNamesExcludingArray ++ Array(s"explode_outer($fieldName1) as $fieldName1")
val explodedDf = df.selectExpr(fieldNamesAndExplode: _*)
lb += fieldName
return getArraysFromDataframe(explodedDf,lb)
case structType: StructType =>
val childFieldnames = structType.fieldNames.map(childname => fieldName + "." + childname)
val newfieldNames = fieldNames.filter(_ != fieldName) ++ childFieldnames
val renamedcols = newfieldNames.map(x => (col(x.toString()).as(x.toString().replace("value.","").replace(".", "_").replace("$", "_").replace("__", "_").replace(" ", "").replace("-", ""))))
val explodedf = df.select(renamedcols: _*)
return getArraysFromDataframe(explodedf,lb)
case _ =>
}
}
lb
}
Example Dataframe/Schema:
val arrayStructSchema = new StructType()
.add("schemaId",StringType)
.add("timestamp",StringType)
.add("partition",IntegerType)
.add("offset",IntegerType)
.add("Key",IntegerType)
.add("value", new StructType()
.add("Entity", new StructType()
.add("EntityId",StringType)
.add("EntityType",StringType)
)
.add("Shipment", new StructType()
.add("ShipmentId", StringType, false)
.add("AttributeDate", StringType, false)
.add("Level2",ArrayType(new StructType()
.add("Level2Id", StringType)
.add("Level2Details", new StructType()
.add("Details", new StructType()
.add("Attribute1", StringType)
.add("Attribute2", StringType))
.add("DetailsType", IntegerType, false))
.add("Level2Attribute1",StringType)
.add("Level2Attribute2",IntegerType)
.add("Level2RefNumbs", ArrayType(new StructType()
.add("RefNumb", StringType)
.add("Attribute2", StringType)
.add("Attribute3", StringType)
.add("Attribute4", StringType))
)
)
)
.add("Level2Order",ArrayType(new StructType()
.add("Level2OrderId", IntegerType)
.add("Level2OrderBillDetails", ArrayType(new StructType()
.add("Level2OrderBillDetailId", StringType)
.add("Level2OrderBillCode", StringType)
.add("Level2OrderRefNumbs", ArrayType(new StructType()
.add("RefNumb", StringType)
.add("Attribute2", StringType)
.add("Attribute3", StringType)
.add("Attribute4", StringType))))
)
)
)//
)//
.add("TestLevel", new StructType()
.add("TestLevelId", StringType, false)
.add("Attribute1", StringType, false)
.add("TestLevel2",ArrayType(new StructType()
.add("TestLevel2Id", StringType)
.add("Attribute2", new StructType()
.add("TestStruct", new StructType()
.add("Attribute2a", StringType)
.add("Attribute2b", StringType))
.add("ServiceLevel", IntegerType, false))
.add("TestLevel3",ArrayType(new StructType()
.add("TestLevel3Id", StringType)
.add("Attribute3", StringType))
)//TestLevel3
)
)//TestLevel2
)//TestLevel
)//value
val df = spark.createDataFrame(spark.sparkContext
.emptyRDD[Row],arrayStructSchema)
Main Code:
val flattenedSchema = flattenSchema(df.schema)
val renamedCols = flattenedSchema.map(name => col(name.toString()).as(name.toString().replace("value.","").replace(".","_")))
var deserializedFilterFlattenedRawDF = df.select(renamedCols:_*)
deserializedFilterFlattenedRawDF.printSchema
var arrayItem = ""
var arrayName = ""
var tableName = ""
var arrayFields = ""
var parentIsArray = false
var parentLevel = ""
var parentId = ""
var parentIdAlias = ""
var listOfArraysInTopic = new ListBuffer[String]()
var dfOfArrays = getArraysFromDataframe(df,listOfArraysInTopic)
for(arrayItem <- dfOfArrays){
print("---->arrayitem:" + arrayItem + "\n")
var arrayPrefix = arrayItem.split("_")
var arrayItemName = arrayPrefix.last
var arrayItemExplode = ""
tableName = arrayItem.substring(arrayItem.lastIndexOf('_') + 1);
arrayName = tableName.concat("Array")
arrayFields = tableName+".*"
var selectColumns = new ListBuffer[String]()
selectColumns += ("Timestamp","Partition","Key","Offset","SchemaId")
selectColumns += arrayFields
println("tableName: "+tableName + "\n")
println("arrayItem: "+arrayItem + "\n")
println("arrayItemName: " + arrayItemName + "\n")
println("arrayName: "+arrayName + "\n")
println("arrayFields: "+arrayFields + "\n")
println("selectColumns: "+selectColumns + "\n")
var dfArray = deserializedFilterFlattenedRawDF
.withColumn(arrayName, explode(array(arrayItem)))
.withColumn(tableName, explode($"${arrayName}"))
.select(selectColumns.map(m=>col(m)):_*)
val flattenedArraySchema = flattenSchema(dfArray.schema)
val renamedArrayCols = flattenedArraySchema.map(name => col(name.toString()).as(name.toString().replace("value.","").replace(".","_")))
var dfArrayparsedFlattened = dfArray.select(renamedArrayCols:_*)
dfArrayparsedFlattened.printSchema
} //end listOfArraysInTopic for loop
The expected results would be a dataframe for each Array: Shipment_Level2, Shipment_Level2Order, TestLevel_TestLevel2, Shipment_Level2_Level2RefNumbs, Shipment_Level2Order_Level2OrderBillDetails, TestLevel_TestLevel2_TestLevel3, Shipment_Level2Order_Level2OrderBillDetails_Level2OrderRefNumbs
And that each generated dataframe now has all the attributes of the array flattened out.