Store Schema of Read File Into csv file in spark scala

user_11742772 0
Viewed 1148

i am reading a csv file using inferschema option enabled in data frame using below command.

df2 = spark.read.options(Map("inferSchema"->"true","header"->"true")).csv("s3://Bucket-Name/Fun/Map/file.csv")

df2.printSchema()

Output:

root
 |-- CC|Fun|Head|Country|SendType: string (nullable = true)

Now I would like to store the above output only into a csv file having just these column names and datatype of these columns like below.

column_name,datatype
CC,string
Fun,string
Head,string
Country,string
SendType,string

I tried writing this into a csv using below option, but this is writing the file with entire data.

df2.coalesce(1).write.format("csv").mode("append").save("schema.csv")

regards mahi

apache-spark csv scala
5 Answers

df.schema.fields to get fields & its datatype.

Check below code.

scala> val schema = df.schema.fields.map(field => (field.name,field.dataType.typeName)).toList.toDF("column_name","datatype")
schema: org.apache.spark.sql.DataFrame = [column_name: string, datatype: string]

scala> schema.show(false)
+---------------+--------+
|column_name    |datatype|
+---------------+--------+
|applicationName|string  |
|id             |string  |
|requestId      |string  |
|version        |long    |
+---------------+--------+


scala> schema.write.format("csv").save("/tmp/schema")
user_85934140

Try something like below use coalesce(1) and .option("header","true") to output with header

import java.io.FileWriter

object SparkSchema {

  def main(args: Array[String]): Unit = {

    val fw = new FileWriter("src/main/resources/csv.schema", true)
    fw.write("column_name,datatype\n")

    val spark = Constant.getSparkSess

    import spark.implicits._

    val df = List(("", "", "", 1l)).toDF("applicationName", "id", "requestId", "version")
    val columnList : List[(String, String)] = df.schema.fields.map(field => (field.name, field.dataType.typeName))
      .toList
    try {
      val outString = columnList.map(col => {
        col._1 + "," + col._2
      }).mkString("\n")
      fw.write(outString)
    }
    finally fw.close()

    val newColumnList : List[(String, String)] = List(("newColumn","integer"))

    val finalColList = columnList ++ newColumnList
    writeToS3("s3://bucket/newFileName.csv",finalColList)

  }

  def writeToS3(s3FileNameWithpath : String,finalColList : List[(String,String)]) {

    val outString =  finalColList.map(col => {
      col._1 + "," + col._2
    }).mkString("\\n")

    import org.apache.hadoop.fs._
    import org.apache.hadoop.conf.Configuration
    val conf = new Configuration()
    conf.set("fs.s3a.access.key", "YOUR ACCESS KEY")
    conf.set("fs.s3a.secret.key", "YOUR SECRET KEY")

    val dest = new Path(s3FileNameWithpath)
    val fs = dest.getFileSystem(conf)
    val out = fs.create(dest, true)
    out.write( outString.getBytes )
    out.close()
  }

}
user_63035790

An alternative to @QuickSilver's and @Srinivas' solutions, which they should both work, is to use the DDL representation of the schema. With df.schema.toDDL you get:

CC STRING, fun STRING, Head STRING, Country STRING, SendType STRING

which is the string representation of the schema then you can split and replace as shown next:

import java.io.PrintWriter

val schema = df.schema.toDDL.split(",")
// Array[String] = Array(`CC` STRING, `fun` STRING, `Head` STRING, `Country` STRING, `SendType` STRING)

val writer = new PrintWriter("/tmp/schema.csv")

writer.write("column_name,datatype\n")
schema.foreach{ r => writer.write(r.replace(" ", ",") + "\n") }
writer.close()

To write to S3 you can use Hadoop API as QuickSilver already implemented or a 3rd party library such as MINIO:

import io.minio.MinioClient

val minioClient = new MinioClient("https://play.min.io", "ACCESS_KEY", "SECRET_KEY")

minioClient.putObject("YOUR_BUCKET","schema.csv", "/tmp/schema.csv", null)

Or even better by generating a string, storing it into a buffer and then send it via InputStream to S3:

import java.io.ByteArrayInputStream
import io.minio.MinioClient

val minioClient = new MinioClient("https://play.min.io", "ACCESS_KEY", "SECRET_KEY")

val schema = df.schema.toDDL.split(",")
val schemaBuffer = new StringBuilder

schemaBuffer ++= "column_name,datatype\n"
schema.foreach{ r => schemaBuffer ++= r.replace(" ", ",") + "\n" }

val inputStream = new ByteArrayInputStream(schemaBuffer.toString.getBytes("UTF-8"))

minioClient.putObject("YOUR_BUCKET", "schema.csv", inputStream, new PutObjectOptions(inputStream.available(), -1))

inputStream.close
user_7503760

@PySpark

df_schema = spark.createDataFrame([(i.name, str(i.dataType)) for i in df.schema.fields], ['column_name', 'datatype'])
df_schema.show()

This will create new dataFrame for schema of existing dataframe
UseCase:

Useful when you want create table with Schema of the dataframe & you cannot use below code as pySpark user may not be authorized to execute DDL commands on database.

df.createOrReplaceTempView("tmp_output_table")
spark.sql("""drop table if exists schema.output_table""")   
spark.sql("""create table schema.output_table as select * from tmp_output_table""")
user_132091010

In Pyspark - You can find all column names & data types (DataType) of PySpark DataFrame by using df.dtypes. Follow this link for more details pyspark.sql.DataFrame.dtypes

Having said that, try using below code -

data = df.dtypes
cols = ["col_name", "datatype"]

df = spark.createDataFrame(data=data,schema=cols)

df.show()
user_151125630
Related