Built with Tutify.io

    AIStudyLabs

    Spark Read ORC file into DataFrame

    Spark Read ORC file into DataFrame

    Spark natively supports ORC data source to read ORC into DataFrame and write it back to the ORC file format using orc() method of DataFrameReader and DataFrameWriter. In this article, I will explain how to read an ORC file into Spark DataFrame, proform some filtering, creating a table by reading the ORC file, and finally writing is back by partition using scala examples.

    Table of contents

    <a id="h-what-is-the-orc-file"></a>

    What is the ORC file?

    ORC stands of Optimized Row Columnar which provides a highly efficient way to store the data in a self-describing, type-aware column-oriented format for the Hadoop ecosystem. This is similar to other columnar storage formats Hadoop supports such as RCFile, parquet.

    ORC file format heavily used as a storage for Apache Hive due to its highly efficient way of storing data which enables high-speed processing and ORC also used or natively supported by many frameworks like Hadoop MapReduce, Apache Spark, Pig, Nifi, and many more.

    <a id="h-orc-advantages"></a>

    ORC Advantages

    <a id="h-orc-compression"></a>

    ORC Compression

    Spark supports the following compression options for ORC data source. By default, it uses SNAPPY when not specified.

    <a id="h-create-a-dataframe"></a>

    Create a DataFrame

    Spark by default supports ORC file formats without importing third party ORC dependencies. Since we don't have an ORC file to read, first will create an ORC file from the DataFrame. Below is a sample DataFrame we use to create an ORC file.

    
val data =Seq(("James ","","Smith","36636","M",3000),
  ("Michael ","Rose","","40288","M",4000),
  ("Robert ","","Williams","42114","M",4000),
  ("Maria ","Anne","Jones","39192","F",4000),
  ("Jen","Mary","Brown","","F",-1))
val columns=Seq("firstname","middlename","lastname","dob","gender","salary")
val df=spark.createDataFrame(data).toDF(columns:_*)
df.printSchema()
df.show(false)

    <a id="h-spark-write-orc-file"></a>

    Spark Write ORC file

    Spark DataFrameWriter uses orc() method to write or create ORC file from DataFrame. This method takes a path as an argument where to write a ORC file.

    
df.write.orc("/tmp/orc/data.orc")

    Alternatively, you can also write using format("orc")

    
df.write.format("orc").save("/tmp/orc/data.orc")

    Spark by default uses snappy compression while writing ORC file. You can notice this on the part file names. And you can change the compression from default snappy to either none or zlib using an option compression

    
  df.write.mode("overwrite")
    .option("compression","zlib")
    .orc("/tmp/orc/data-zlib.orc")

    This creates ORC files with zlib compression.

    Using append save mode, you can append a DataFrame to an existing ORC file. Incase to overwrite use overwrite save mode.

    
df.write.mode('append').orc("/tmp/orc/people.orc")
df.write.mode('overwrite').orc("/tmp/orc/people.orc")

    <a id="h-spark-read-orc-file"></a>

    Spark Read ORC file

    Use Spark DataFrameReader's orc() method to read ORC file into DataFrame. This supports reading snappy, zlib or no compression, it is not necessary to specify in compression option while reading a ORC file.

    
df.read.orc("/tmp/orc/data.orc")

    In order to read ORC files from Amazon S3, use the below prefix to the path along with third-party dependencies and credentials.

    <a id="h-executing-sql-queries-on-dataframe"></a>

    Executing SQL queries on DataFrame

    We can also create a temporary view on Stark DataFrame that was created on ORC file and run SQL queries.. These views are available until your program exits.

    
df2.createOrReplaceTempView("ORCTable")
val orcSQL = spark.sql("select firstname,dob from ORCTable where salary >= 4000 ")
orcSQL.show(false)

    In this example, the physical table scan loads only columns firstname, dob, and age at runtime, without reading allcolumns from the file system. This improves read performance.

    <a id="h-creating-a-table-on-orc-file"></a>

    Creating a table on ORC file

    Now let’s walk through executing SQL queries on the ORC file without creating a DataFrame first. In order to execute SQL queries, create a temporary view or table directly on the ORC file instead of creating from DataFrame.

    
spark.sql("CREATE TEMPORARY VIEW PERSON USING orc OPTIONS (path \"/tmp/orc/data.orc\")")
spark.sql("SELECT * FROM PERSON").show()

    Here, we created a temporary view PERSON from ORC file “data” file. This gives the following results.

    
+---------+----------+--------+-----+------+------+
|firstname|middlename|lastname|  dob|gender|salary|
+---------+----------+--------+-----+------+------+
|  Robert |          |Williams|42114|     M|  4000|
|   Maria |      Anne|   Jones|39192|     F|  4000|
| Michael |      Rose|        |40288|     M|  4000|
|   James |          |   Smith|36636|     M|  3000|
|      Jen|      Mary|   Brown|     |     F|    -1|
+---------+----------+--------+-----+------+------+

    <a id="h-using-partition"></a>

    Using Partition

    When we execute a particular query on PERSON table, it scan's through all the rows and returns the results the selected columns back. In Spark, we can improve query execution in an optimized way by doing partitions on the data using partitionBy() method. Following is the example of partitionBy().

    
df.write.partitionBy("gender","salary")
    .mode("overwrite").orc("/tmp/orc/data.orc")

    When you check the people.orc file, it has two partitions "gender" followed by "salary" inside.

    <a id="h-reading-a-specific-partition"></a>

    Reading a specific Partition

    The example below explains of reading partitioned ORC file into DataFrame with gender=M.

    
val parDF=spark.read.orc("/tmp/orc/data.orc/gender=M")
parDF.show(false)

    <a id="h-which-compression-to-choose"></a>

    Which compression to choose

    Not writing ORC files in compression results in larger disk space and slower in performance. Hence, it is suggestable to use compression. Below are basic comparison between ZLIB and SNAPPY when to use what.

    <a id="h-complete-example-of-using-orc-in-spark"></a>

    Complete Example of using ORC in Spark

    
import org.apache.spark.sql.{SparkSession}

object ReadORCFile extends App{

  val spark: SparkSession = SparkSession.builder()
    .master("local[1]")
    .appName("AIStudyLabs.com")
    .getOrCreate()

  val data =Seq(("James ","","Smith","36636","M",3000),
  ("Michael ","Rose","","40288","M",4000),
  ("Robert ","","Williams","42114","M",4000),
  ("Maria ","Anne","Jones","39192","F",4000),
  ("Jen","Mary","Brown","","F",-1))
  val columns=Seq("firstname","middlename","lastname","dob","gender","salary")
  val df=spark.createDataFrame(data).toDF(columns:_*)

  df.write.mode("overwrite")
    .orc("/tmp/orc/data.orc")

  df.write.mode("overwrite")
    .option("compression","none12")
    .orc("/tmp/orc/data-nocomp.orc")

  df.write.mode("overwrite")
    .option("compression","zlib")
    .orc("/tmp/orc/data-zlib.orc")

  val df2=spark.read.orc("/tmp/orc/data.orc")
  df2.show(false)

  df2.createOrReplaceTempView("ORCTable")
  val orcSQL = spark.sql("select firstname,dob from ORCTable where salary >= 4000 ")
  orcSQL.show(false)

  spark.sql("CREATE TEMPORARY VIEW PERSON USING orc OPTIONS (path \"/tmp/orc/data.orc\")")
  spark.sql("SELECT * FROM PERSON").show()
}

    <a id="h-conclusion"></a>

    Conclusion

    In summary, ORC is a high efficient, compressed columnar format that is capable to store petabytes of data without compromising fast reads. Spark natively supports ORC data source to read and write an ORC files using orc() method on DataFrameReader and DataFrameWrite.

    <a id="h-related-articles"></a>

    Related Articles

    <a id="h-reference"></a>

    Reference