मेरे पास एक नमूना डेटाफ्रेम है जिसे मैं नीचे दिए गए कोड का उपयोग करके बनाता हूं:
val data = Seq(
Row(20.0, "dog"),
Row(3.5, "cat"),
Row(0.000006, "ant")
)
val schema = StructType(
List(
StructField("weight", DoubleType, true),
StructField("animal_type", StringType, true)
)
)
val df = spark.createDataFrame(
spark.sparkContext.parallelize(data),
schema
)
val actualDF = df.withColumn(
"animal_interpretation",
struct(
(col("weight") > 5).as("is_large_animal"),
col("animal_type").isin("rat", "cat", "dog").as("is_mammal")
)
)
actualDF.show(false)
+------+-----------+---------------------+
|weight|animal_type|animal_interpretation|
+------+-----------+---------------------+
|20.0 |dog |[true,true] |
|3.5 |cat |[false,true] |
|6.0E-6|ant |[false,false] |
+------+-----------+---------------------+
इस स्पार्क डीएफ की स्कीमा का उपयोग करके मुद्रित किया जा सकता है -
scala> actualDF.printSchema
root
|-- weight: double (nullable = true)
|-- animal_type: string (nullable = true)
|-- animal_interpretation: struct (nullable = false)
| |-- is_large_animal: boolean (nullable = true)
| |-- is_mammal: boolean (nullable = true)
हालांकि, मैं इस स्कीमा को डेटाफ्रेम के रूप में प्राप्त करना चाहता हूं जिसमें 3 कॉलम हैं - field, type, nullable। स्कीमा से आउटपुट डेटाफ्रेम कुछ इस तरह होगा -
+-------------------------------------+--------------+--------+
|field |type |nullable|
+-------------------------------------+--------------+--------+
|weight |double |true |
|animal_type |string |true |
|animal_interpretation |struct |false |
|animal_interpretation.is_large_animal|boolean |true |
|animal_interpretation.is_mammal |boolean |true |
+----------------------------------------------------+--------+
मैं इसे स्पार्क में कैसे प्राप्त कर सकता हूं। मैं कोडिंग के लिए स्कैला का उपयोग कर रहा हूं।
2 जवाब
यहां आपके कोड सहित एक पूरा उदाहरण दिया गया है। मैंने मिलान के लिए कुछ सामान्य फ़्लैटनस्केमा विधि का उपयोग किया जैसे शंकर ने स्ट्रक्चर को पार करने के लिए किया था, लेकिन इस विधि को फ़्लैटेड स्कीमा वापस करने के बजाय मैंने स्ट्रक्चर टाइप के डेटाटाइप को एकत्र करने के लिए एक ऐरेबफर का उपयोग किया और ऐरेबफर को वापस कर दिया। मैंने तब ArrayBuffer को एक सीक्वेंस में बदल दिया और अंत में, स्पार्क का उपयोग करके, सीक्वेंस को एक DataFrame में बदल दिया।
import org.apache.spark.sql.types.{StructType, StructField, DoubleType, StringType}
import org.apache.spark.sql.functions.{struct, col}
import scala.collection.mutable.ArrayBuffer
val data = Seq(
Row(20.0, "dog"),
Row(3.5, "cat"),
Row(0.000006, "ant")
)
val schema = StructType(
List(
StructField("weight", DoubleType, true),
StructField("animal_type", StringType, true)
)
)
val df = spark.createDataFrame(
spark.sparkContext.parallelize(data),
schema
)
val actualDF = df.withColumn(
"animal_interpretation",
struct(
(col("weight") > 5).as("is_large_animal"),
col("animal_type").isin("rat", "cat", "dog").as("is_mammal")
)
)
var fieldStructs = new ArrayBuffer[(String, String, Boolean)]()
def flattenSchema(schema: StructType, fieldStructs: ArrayBuffer[(String, String, Boolean)], prefix: String = null): ArrayBuffer[(String, String, Boolean)] = {
schema.fields.foreach(field => {
val col = if (prefix == null) field.name else (prefix + "." + field.name)
field.dataType match {
case st: StructType => {
fieldStructs += ((col, field.dataType.typeName, field.nullable))
flattenSchema(st, fieldStructs, col)
}
case _ => {
fieldStructs += ((col, field.dataType.simpleString, field.nullable))
}
}}
)
fieldStructs
}
val foo = flattenSchema(actualDF.schema, fieldStructs).toSeq.toDF("field", "type", "nullable")
foo.show(false)
यदि आप उपरोक्त चलाते हैं तो आपको निम्न प्राप्त करना चाहिए।
+-------------------------------------+-------+--------+
|field |type |nullable|
+-------------------------------------+-------+--------+
|weight |double |true |
|animal_type |string |true |
|animal_interpretation |struct |false |
|animal_interpretation.is_large_animal|boolean|true |
|animal_interpretation.is_mammal |boolean|true |
+-------------------------------------+-------+--------+
आप कुछ ऐसा कर सकते हैं
def flattenSchema(schema: StructType, prefix: String = null) : Seq[(String, String, Boolean)] = {
schema.fields.flatMap(field => {
val col = if (prefix == null) field.name else (prefix + "." + field.name)
field.dataType match {
case st: StructType => flattenSchema(st, col)
case _ => Array((col, field.dataType.simpleString, field.nullable))
}
})
}
flattenSchema(actualDF.schema).toDF("field", "type", "nullable").show()
उम्मीद है ये मदद करेगा!