{"action":"created","comment":{"url":"https://api.github.com/repos/databricks/spark-redshift/pulls/comments/40826790","id":40826790,"diff_hunk":"@@ -0,0 +1,413 @@\n+#Tutorial#\n+\n+The [Spark Data Sources API](https://databricks.com/blog/2015/01/09/spark-sql-data-sources-api-unified-data-access-for-the-spark-platform.html) introduced in Spark 1.2 supports a pluggable mechanism for integration with structured data-sources. It is a unified API designed to support two major operations:\n+\n+1. Loading structured data from an external data source into Spark\n+2. Storing structured data from Spark into an external data source.\n+\n+The Data Sources API has built in integration for several data sources such as Hive, Avro, Json, JDBC and Parquet. Third party integration is added through spark-packages. One such integration is for Amazon Redshift which is provided by the spark-redshift package. \n+\n+Prior to the introduction of spark-redshift, Spark’s JDBC data source was the only way for Spark users to read data from Redshift. While this method is adequate when running queries returning a small number of rows (order of 100’s), it is too slow when handling large scale data. The reason being that JDBC provides a ResultSet based approach where rows are retrieved in a single thread in small batches. Furthermore, the use of JDBC to store large datasets in Redshift is only practical when data needs to be moved between tables inside a Redshift database. The JDBC based INSERT/UPDATE queries are only practical for small updates to Redshift tables. For users hoping to load or store large volumes of data from Redshift, JDBC leaves much to be desired in terms of performance and throughput.\n+\n+This tutorial will provide a hand-on experience in using the spark-redshift package from your local development environment. It will also provide a deep dive into the implementation details of spark-redshift which will enable you to gain a deeper understanding of why spark-redshift provides a high performance alternative to plain JDBC based approach to interacting with Redshift from Spark.\n+\n+##Prepare the Redshift database##\n+\n+Before we delve into specific examples of how spark-redshift works let us configure the Redshift database which we will be using.\n+\n+For the purpose of this article I will use the sample [TICKT](http://docs.aws.amazon.com/redshift/latest/dg/c_sampledb.html) database on Redshift. This database enables the tracking of  sales activity for the fictional TICKIT web site, where users buy and sell tickets online for various types of events. The database allows analysis of sales over time, performance of sellers, venues and correlation of sales with seasons. This information can be used to drive advertising and promotions campaigns.  \n+\n+When you start the Redshift service you will first need to create the TICKT database and load it. Follow the instructions [here](http://docs.aws.amazon.com/redshift/latest/dg/cm-dev-t-load-sample-data.html) to create/load the TICKT database.\n+\n+I used a Redshift database running on a 2 node cluster. Each node manages 2 slices for a total of 4 slices. This *usually* (read Redshift [documentation](http://docs.aws.amazon.com/redshift/latest/dg/c_high_level_system_architecture.html) for a more nuanced discussion) means that each table will be stored in 4 separate partitions, one for each slice. \n+\n+## Usage##\n+\n+We are ready to interact with Redshift using the spark-redshift library. The skeleton of the program we will be using is shown in Listing 1. The entire `SparkRedshiftTutorial.scala` program can be accessed from [here](SparkRedshiftTutorial.scala). You can also use the Spark REPL to run the lines listed in the program below. \n+\n+```scala\n+package com.databricks.spark.redshift.tutorial\n+import org.apache.spark.{SparkConf,SparkContext}\n+import org.apache.spark.sql.SaveMode\n+import java.io.File\n+import org.apache.spark.sql.types.{StructType,StructField,StringType,IntegerType,LongType,DecimalType}\n+\n+\n+object SparkRedshiftTutorial {\n+  /*\n+   * For Windows Users only\n+   * 1. Download contents from link https://github.com/srccodes/hadoop-common-2.2.0-bin/archive/master.zip\n+   * 2. Unzip the file in step 1 into your %HADOOP_HOME%/bin. In the example below  %HADOOP_HOME%=C:/MySoftware/hdphome\n+   * 3. After step 2 %HADOOP_HOME/bin should contain winutils.exe\n+   * \n+  */\n+  if(System.getProperty(\"os.name\").startsWith(\"Windows\")){\n+    val hdphome:File = new File(\"C:/MySoftware/hdphome\");\n+    System.setProperty(\"hadoop.home.dir\",hdphome.getAbsolutePath().toString())    \n+  }\n+  \n+  def main(args:Array[String]):Unit = {\n+    val awsAccessKeyId = args(0) \n+    val awsSecretAccessKey = args(1)    \n+    val redshiftDBName = args(2)\n+    val redshiftUserId= args(3)\n+    val redshiftPassword= args(4)\n+    /**\n+      * Sample URL-swredshift.czac2vcs84ci.us-east-1.redshift.amazonaws.com:5439\n+      */\n+    val redshifturl = args(5) \n+    val jdbcURL = s\"jdbc:redshift://$redshifturl/$redshiftDBName?user=$redshiftUserId&password=$redshiftPassword\"\n+\n+    val sc = new SparkContext(new SparkConf().setAppName(\"SparkSQL\").setMaster(\"local\"))\n+\n+\t/*Configure SparkContext to communicate with AWS*/\n+\tval tempS3Dir = \"s3n://redshift-spark/temp/\"\n+    sc.hadoopConfiguration.set(\"fs.s3n.awsAccessKeyId\", awsAccessKeyId)\n+    sc.hadoopConfiguration.set(\"fs.s3n.awsSecretAccessKey\", awsSecretAccessKey)\n+\n+\t/*Create the SQL Context*/    \n+    val sqlContext = new org.apache.spark.sql.SQLContext(sc)\n+  }\n+} \n+```\n+Listing 1 : SparkRedshiftExamples.scala skeleton\n+\n+We need the following set of user provided parameters to communicate with AWS in general and Redshift in particular -\n+\n+- **AWS Access Key and AWS Secret Access Key** - This key pair will be used to communicate with AWS services. This information is passed by the AWS Client libraries in every interaction with AWS.\n+- **Redshift Database Name** - When you provision the Redshift service you have to provide a name of for your database. This is similar to a schema in Oracle. The name of my Redshift database was sparkredshift\n+- **Redshift UserId/Password combination** - You will need to provide this information when the Redshift service was provisioned.\n+- **Redshift URL** - You will need to obtain this from your Redshift Console. A sample Redshift URL is `swredshift.czac2vcs84ci.us-east-1.redshift.amazonaws.com:5439`\n+\n+Your Redshift console will provide the JDBC URL to use. It follows the pattern\n+\n+`jdbc:redshift://$redshifturl/$redshiftDBName?user=$redshiftUserId&password=$redshiftPassword`\n+\n+A sample JDBC URL is\n+\n+`jdbc:redshift://swredshift.czac2vcs84ci.us-east-1.redshift.amazonaws.com:5439/sparkredshift?user=spark&password=mysecretpass`\n+\n+spark-redshift reads and writes data to S3 when transferring data to/from Redshift, so you'll need to specify a path in S3 where the library should write these temporary files. spark-redshift cannot automatically clean up the temporary files it creates in S3. As a result, we recommend that you use a dedicated temporary S3 bucket with an [object lifecycle configuration ](http://docs.aws.amazon.com/AmazonS3/latest/dev/object-lifecycle-mgmt.html) to ensure that temporary files are automatically deleted after a specified expiration period. For this example we create a S3 bucket `redshift-spark`. We tell spark-redshift that we will use the following temporary location in S3 to store temporary files generated by spark-redshift `s3n://redshift-spark/temp/`\n+\n+Next enable the communication with AWS by configuring the following properties in the `SparkContext` instance-\n+\n+- fs.s3n.awsAccessKeyId \n+- fs.s3n.awsSecretAccessKey\n+\n+Lastly we create the SQLContext to use the Data Source API to communicate with Redshift\n+\n+`val sqlContext = new org.apache.spark.sql.SQLContext(sc)`\n+\n+### Load Function - Reading from a Redshift table###\n+\n+Let us fetch data from the Redshift table `event`. Add the following lines of code listed in Listing 2 to the program in Listing 1\n+\n+```scala\n+import sqlContext.implicits._\n+val eventsDF = sqlContext.read\n+               .format(\"com.databricks.spark.redshift\")\n+               .option(\"url\",jdbcURL ) \n+               .option(\"tempdir\", tempS3Dir) \n+               .option(\"dbtable\", \"event\")\n+               .load()\n+eventsDF.show()   \n+```\n+Listing 2 : Reading an entire Redshift table \n+\n+`.format(\"com.databricks.spark.redshift\")` line tells the DataSource API that we are using the `spark-redshift` package. It uses this information to identify the class `DefaultSource` in the package specified by the `format` invocation. This class contains the entry points for the functionality provided by the Data Sources API implementation.\n+\n+Next we provide the parameters necessary to read the `event` table. We provide the JDBC URL, the temporary S3 folder which will be used to copy the table data to and the table name we want to read. A comprehensive list of parameters is listed on the spark-redshift documentation [page](https://github.com/databricks/spark-redshift#parameters).\n+\n+The lines in Listing 2 will execute to produce the following output\n+\n+```{r, engine='bash'}\n++-------+-------+-----+------+------------------+--------------------+\n+|eventid|venueid|catid|dateid|         eventname|           starttime|\n++-------+-------+-----+------+------------------+--------------------+\n+|   1433|    248|    6|  1827|            Grease|2008-01-01 19:00:...|\n+|   2811|    207|    7|  1827|  Spring Awakening|2008-01-01 15:00:...|\n+|   4135|     16|    9|  1827|               Nas|2008-01-01 14:30:...|\n+|   5807|     45|    9|  1827| Return To Forever|2008-01-01 15:00:...|\n+|   1738|    260|    6|  1828|      Beatles LOVE|2008-01-02 20:00:...|\n+|   2131|    212|    7|  1828|           Macbeth|2008-01-02 15:00:...|\n+|   2494|    203|    7|  1828|     The Caretaker|2008-01-02 19:00:...|\n+|   2824|    248|    7|  1828|  Cirque du Soleil|2008-01-02 19:00:...|\n+|   2920|    209|    7|  1828|           Macbeth|2008-01-02 19:30:...|\n+|   4853|     39|    9|  1828|        Rick Braun|2008-01-02 19:00:...|\n+|    394|    300|    8|  1829|Adriana Lecouvreur|2008-01-03 15:00:...|\n+|   2043|    217|    7|  1829|       The Bacchae|2008-01-03 19:00:...|\n+|   5508|     55|    9|  1829|        Ryan Adams|2008-01-03 19:30:...|\n+|   6071|     47|    9|  1829|      3 Doors Down|2008-01-03 15:00:...|\n+|   6120|    130|    9|  1829|         Bob Dylan|2008-01-03 19:30:...|\n+|   7468|     37|    9|  1829|        Commodores|2008-01-03 19:00:...|\n+|   1567|    257|    6|  1830|    Blue Man Group|2008-01-04 15:00:...|\n+|   1764|    262|    6|  1830|        Mamma Mia!|2008-01-04 20:00:...|\n+|   1981|    219|    7|  1830|         King Lear|2008-01-04 15:00:...|\n+|   2274|    222|    7|  1830|     The Caretaker|2008-01-04 20:00:...|\n++-------+-------+-----+------+------------------+--------------------+\n+only showing top 20 rows\n+```\n+\n+spark-redshift automatically reads the schema from the Redshift table and maps its types back to Spark SQL's types. The command `eventsDF.printSchema()` produces the following output\n+\n+```{r, engine='bash'}\n+root\n+ |-- eventid: integer (nullable = true)\n+ |-- venueid: integer (nullable = true)\n+ |-- catid: integer (nullable = true)\n+ |-- dateid: integer (nullable = true)\n+ |-- eventname: string (nullable = true)\n+ |-- starttime: timestamp (nullable = true)\n+```\n+\n+We can even register the DataFrame instance as a temporary table in Spark and execute queries against it as follows\n+\n+```scala\n+eventsDF.registerTempTable(\"myevent\")\n+val myEventDF = sqlContext.sql(\"select * myevent\")\n+```\n+\n+While the above examples used Scala we could have also used SQL. An example of writing to a Redshift table in SQL is show below\n+\n+```{r, engine='bash'}\n+> %sql\n+\n+CREATE TEMPORARY TABLE myevent \n+USING com.databricks.spark.redshift\n+OPTIONS (\n+  dbtable 'event', \n+  tempdir 's3n://redshift-spark/temp/',\n+  url 'jdbc:redshift://swredshift.czac2vcs84ci.us-east-1.redshift.amazonaws.com:5439/sparkredshift?user=spark&password=mysecretpass'\n+);     \n+select * from myevent; \n+```\n+\n+Note that we have registered a temporary table `myevent` in Spark and executed a query against it (`select * from myevent`) just like we did in our Scala example.\n+\n+### Load Function - Reading from a Redshift query###\n+\n+We can also read from a Redshift query. Listing 3 shows how to read the most recent 10000 records from the `sales` table. Notice that the line `.option(\"dbtable\", \"event\")` is replaced with the line `.option(\"query\", salesQuery) `\n+\n+```scala\n+val salesQuery = \"\"\"select salesid,listid,sellerid,buyerid,\n+                           eventid,dateid,qtysold,pricepaid,commission \n+                           from sales \n+                           order by saletime desc LIMIT 10000\"\"\"\n+val salesDF = sqlContext.read\n+             .format(\"com.databricks.spark.redshift\")\n+             .option(\"url\",jdbcURL ) \n+             .option(\"tempdir\", tempS3Dir) \n+             .option(\"query\", salesQuery)\n+             .load()\n+salesDF.show()\n+```\n+Listing 3: Reading from a Redshift query\n+\n+The lines in Listing 3 will execute to produce the following output\n+\n+```{r, engine='bash'}\n++-------+------+--------+-------+-------+------+-------+---------+----------+\n+|salesid|listid|sellerid|buyerid|eventid|dateid|qtysold|pricepaid|commission|\n++-------+------+--------+-------+-------+------+-------+---------+----------+\n+| 102373|117133|   26190|  35491|   2141|  2191|      4|  1008.00|    151.20|\n+|  75861| 86640|   43402|  39545|   8372|  2191|      2|   372.00|     55.80|\n+|  56302| 63546|    5760|   5797|   1489|  2191|      2|   372.00|     55.80|\n+|  83603| 95341|   27027|  27881|   6034|  2191|      1|   288.00|     43.20|\n+|  40652| 45468|   27557|  28366|   5099|  2191|      2|   510.00|     76.50|\n+| 157586|206655|   13213|  47394|   3665|  2191|      2|  4018.00|    602.70|\n+|  99761|114026|    7466|  21189|   5621|  2191|      2|   102.00|     15.30|\n+|  68467| 77797|   49538|  27029|   1489|  2191|      2|   188.00|     28.20|\n+|  66470| 75420|   22851|  37849|   7604|  2191|      2|   420.00|     63.00|\n+|  36700| 40838|   15099|  45514|   8027|  2191|      4|  1400.00|    210.00|\n+| 126896|145346|   18243|  10273|   6783|  2191|      2|   866.00|    129.90|\n+|  55416| 62529|   42223|  26273|   1489|  2191|      4|   532.00|     79.80|\n+|  48503| 54812|   42828|  12282|   8372|  2191|      2|    80.00|     12.00|\n+|   3737|  4035|   11666|  23692|   6034|  2191|      2|   302.00|     45.30|\n+| 154069|178359|    3897|   5815|   8168|  2191|      1|   290.00|     43.50|\n+|  37817| 42124|   24592|  26543|   6783|  2191|      4|  1300.00|    195.00|\n+|    924|   933|   48898|  25710|   1594|  2191|      3|   504.00|     75.60|\n+| 112232|128585|   30390|  44375|    914|  2191|      1|   126.00|     18.90|\n+|  97099|110826|   24730|  32758|   3331|  2190|      1|   319.00|     47.85|\n+|  74902| 85497|   45049|   2318|   7547|  2190|      2|   554.00|     83.10|\n++-------+------+--------+-------+-------+------+-------+---------+----------+\n+only showing top 20 rows\n+```\n+\n+### Under the hood - Load Function ###\n+\n+In this section we will take a peek inside spark-redshift to understand how exactly the LOAD function works. Specifically we will look at how `event` table was read into a DataFrame. The LOAD is a two step process\n+\n+1. [UNLOAD](http://docs.aws.amazon.com/redshift/latest/dg/r_UNLOAD.html) data from the Redshift table to S3\n+2. Consume the unloaded files in S3 via a custom InputFormat into an RDD which is then wrapped as a DataFrame instance using the schema obtained from Redshift.\n+\n+#### UNLOAD Redshift to S3 ####\n+\n+Figure 1 shows the the actions required to perform this step.\n+\n+![](images/loadunloadstep.gif)\n+\n+Figure 1 : UNLOAD action\n+\n+First the Spark Driver communicates with the Redshift Leader node to obtain the schema of the table (or query) requested. The attribute `override lazy val schema: StructType` in the class `com.databricks.spark.redshift.RedshiftRelation` will obtain the schema on demand by invoking the method `resolveTable` of the class `com.databricks.spark.redshift.JDBCWrapper`. The `JDBCWrapper` class is responsible for fetching the schema from the Redshift Leader.   \n+\n+Next a Redshift [UNLOAD](http://docs.aws.amazon.com/redshift/latest/dg/r_UNLOAD.html) query is created using the schema information obtained. The UNLOAD command unloads each slice into a s3 folder (22c365b4-13cb-40fd-b4d6-d0ac5d426551) created in the temporary s3 location (s3n://spark-redshift/temp/) provided by the user. Each file contains a row per line and each column of the row is pipe (|) delimited. This process occurs in parallel for each slice. The spark-redshift library achieves its high performance through this mechanism.\n+\n+#### Read UNLOAD'ed S3 files into a DataFrame instance####\n+\n+Figure 2 shows how the files unloaded in S3 are consumed to form a DataFrame instance\n+\n+![](images/loadreadstep.gif)\n+Figure 2 : Read from S3 into a DataFrame  \n+\n+Once the files are written to S3, a custom InputFormat implemented in the class `com.databricks.spark.redshift.RedshiftInputFormat` is used to consume the files in parallel. This class is similar to the standard and well known `TextInputFormat` where the key is the byte offset of the start of each line in the file. The value class however, is of type `Array[String]` (unlike `TextInputFormat`where it is `Text`). The value instance is created by splitting the lines using the default delimiter |. The `RedshiftInputFormat` consumes the S3 files line by line to produce an `RDD`. The schema obtained earlier is applied on this `RDD` to generate an instance of a `DataFrame`.\n+\n+### Save Function - Writing to a Redshift table###\n+\n+spark-redshift allows you to write data back to Redshift. The source data is a `DataFrame` instance. We will need to assume that we have a source table available in the environment. This table can be sourced could be from a variety of sources such as a Hive table, CSV file, parquet file, delimited text file, etc. We will source a DataFrame from Redshift and write it back. This is for illustration purposes only. In practice the source would not be Redshift since in such a case it is easiest to perform a \"CREATE TABLE AS\" query on Redshift without going through Spark-SQL. \n+\n+Listing 4 will register the `event` table as `myevent` table in the Spark environment.\n+\n+```scala\n+val eventQuery = \"select * from event\"\n+val eventDF = sqlContext.read\n+             .format(\"com.databricks.spark.redshift\")\n+             .option(\"url\",jdbcURL ) \n+             .option(\"tempdir\", tempS3Dir)\n+             .option(\"query\", eventQuery) \n+             .load()   \n+eventsDF.registerTempTable(\"myevent\") \n+```\n+Listing 4: Create a table myevent in the Spark environment\n+\n+Listing 5 demonstrates how to write the contents of `myevent` to a Redshift table `redshiftevent`\n+\n+```scala\n+/*\n+ *Create a new table redshiftevent after dropping any existing redshiftevent table\n+ *and write event records with event id less than 1000\n+ */\n+sqlContext.sql(\"select * from myevent where eventid<=1000\").withColumnRenamed(\"eventid\", \"id\")\n+             .write.format(\"com.databricks.spark.redshift\")\n+            .option(\"url\", jdbcURL)\n+            .option(\"tempdir\", tempS3Dir)\n+            .option(\"dbtable\", \"redshiftevent\")\n+            .mode(SaveMode.Overwrite) \n+            .save()\n+/*\n+ *Append to an existing table redshiftevent if it exists or create a new one if it does not\n+ *exist and write event records with event id greater than 1000\n+ */\n+\n+sqlContext.sql(\"select * from myevent where eventid>1000\").withColumnRenamed(\"eventid\", \"id\")\n+          .write.format(\"com.databricks.spark.redshift\")\n+          .option(\"url\", jdbcURL) \n+          .option(\"tempdir\", tempS3Dir) \n+          .option(\"dbtable\", \"redshiftevent\") \n+          .mode(SaveMode.Append)\n+          .save()\n+```\n+Listing 5: Writing to Redshift tables\n+\n+spark-redshift automatically creates a Redshift table with the appropriate schema determined from the table/DataFrame being written. The default behavior is to create a new table and to throw an error message if a table with the same name already exists (corresponding to `SaveMode.ErrorIfExists`)\n+\n+There are two key points to note in Listing 5-\n+\n+1. Note the `.withColumnRenamed(\"eventid\", \"id\")`. This feature if necessary if any of the source table columns names are key words in Redshift (ex. table) then spark-redshift will throw and error. This will be [fixed](https://github.com/databricks/spark-redshift/issues/80) in the next release \n+2. Note how we utilize the modes. The first write utilizes the mode `SaveMode.Overwrite` which means that the table will be dropped and recreated it exists. The second query utilizes `SaveMode.Append` which will create the table if it does not exist but will append to it if it does. The default mode is `SaveMode.ErrorIfExists` which creates the table if it does not exists and throws and error if it does. The last mode is `SaveMode.Ignore` which is same as `SaveMode.Overwrite` if the table does not exist but does nothing if the table exists.\n+\n+While the above examples used Scala we could have also used SQL. And example of writing to a Redshift table in SQL is shows in Listing 6. The only thing to be aware of in the SQL mode is that SaveMode's are not available and only the default mode (`SaveMode.ErrorIfExists`) is in force.\n+\n+```sql\n+> %sql\n+CREATE TABLE redshiftevent\n+USING com.databricks.spark.redshift\n+OPTIONS (\n+dbtable 'redshiftevent',\n+tempdir 's3n://redshift-spark/temp/',\n+  url 'jdbc:redshift://swredshift.czac2vcs84ci.us-east-1.redshift.amazonaws.com:5439/sparkredshift?user=spark&password=mysecretpass'\n+)\n+AS SELECT * FROM myevent;\n+```\n+\n+By default the save operation uses the key distribution [style](http://docs.aws.amazon.com/redshift/latest/dg/c_choosing_dist_sort.html) of `EVEN` in Redshift. This can be changed by the use of option parameter `diststyle` and `distkey`. See spark-redshift [documentation](https://github.com/databricks/spark-redshift) for details.\n+\n+### Under the hood - Save Function ###\n+\n+The implementation of the Save function is provided in the class `com.databricks.spark.redshift.RedshiftWriter`.   \n+\n+![](images/savetoredshift.gif)\n+\n+The sequence of steps involved in performing the save function are-\n+\n+1. The data from the `DataFrame` instance is written out in parallel to the temporary S3 folder. spark-avro is used to write the data to ensure that it encapsulates the schema.\n+2. Next a Redshift [COPY](http://docs.aws.amazon.com/redshift/latest/dg/r_COPY.html) command is created and invoked to load the files created in the temporary S3 folder are loaded it into a Redshift table which respects the value of the `SaveMode` parameter passed.\n+\n+\n+## Integration multiple Datasources ##\n+\n+Data read via spark-redshift is automatically converted to `DataFrame` objects, Spark’s primary abstraction for large datasets. This promotes interoperability between data sources, since types are automatically converted to Spark’s standard representations(for example `StringType`, `DecimalType`). A Redshift user can, for instance, join Redshift tables with data stored in S3, Hive tables, CSV or Parquet files stored on HDFS. This flexibility is important to users with complex data pipelines involving multiple sources.\n+\n+To illustrate this concepts consider Listing 6\n+\n+```scala\n+ \n+val salesAGGQuery = \"\"\"select sales.eventid as id,\n+                       sum(qtysold) as totalqty,sum(pricepaid) as salesamt\n+                       from sales\n+                       group by (sales.eventid)\n+                   \"\"\"\n+val salesAGGDF = sqlContext.read\n+               .format(\"com.databricks.spark.redshift\")\n+               .option(\"url\",jdbcURL)\n+               .option(\"tempdir\", tempS3Dir)\n+               .option(\"query\", salesAGGQuery)\n+               .load()   \n+\n+   \n+val salesAGGDF2=salesAGGDF.join(eventsDF, salesAGGDF(\"id\") === eventsDF(\"eventid\"))\n+                         .select(\"id\", \"eventname\",\"totalqty\",\"salesamt\")\n+```\n+Listing 6: Working with DataFrame instances\n+\n+While both `eventDF` and `salesAGGDF` table are sourced from Redshift, assume for the sake of illustration that they are sourced from a non-Redshift datasource. This should not be too far-fetched since the DataFrame class is part of the Data Sources API and not the spark-redshift library. A DataFrame can be created from any data source compatible with the Data Sources API\n+\n+`salesAGGDF2` DataFrame is created by joining `eventsDF` and `salesAGGDF2` where we register `salesAGGDFWithSchema` with a custom schema and register it as a temporary table `redshift_sales_agg` before saving it to Redshift with name  `redshift_sales_agg`\n+  \n+Next consider Listing 7\n+\n+```scala\n+/*Provide a custom schema*/\n+val schema = StructType(List(StructField(\"eventid\",IntegerType,false),\n+                                StructField(\"eventname\",StringType,false),\n+                                StructField(\"qtysold\",LongType,false),\n+                                StructField(\"revenue\",DecimalType(38,2),false)\n+                                ))\n+val salesAGGDFWithSchema = sqlContext.createDataFrame(salesAGGDF2.rdd, schema)\n+salesAGGDFWithSchema.registerTempTable(\"redshift_sales_agg\")\n+\n+sqlContext.sql(\"select * from redshift_sales_agg\")\n+             .write.format(\"com.databricks.spark.redshift\")\n+             .option(\"url\", jdbcURL) \n+             .option(\"tempdir\", tempS3Dir) \n+             .option(\"dbtable\", \"redshift_sales_agg\") \n+             .mode(SaveMode.Overwrite)\n+             .save()\n+```\n+Listing 7: Providing Custom Schema\n+\n+Note how we have specified every column as non-nullable in the custom schema.\n+\n+In this section we illustrated how Data Sources API provides a unified API to work with multiple datasources. This feature is enabled by the generic `DataFrame` interface. We also looked at how a `DataFrame` can be saved to Redshift with a user-defined schema. None of this is unique to spark-redshift. It is provided by the DataSources API for use with all compatible packages. This is the feature that makes the DataSource API such a powerful tool for ETL usecases.\n+\n+## Under the hood - Putting it all together##\n+\n+As we discussed earlier Spark SQL will introspect for a class called `DefaultSource` in the Datasources API package `com.databricks.spark.redshift`. The `DefaultSource` class implements the trait `RelationProvider` class which is provides the default load functionality for the library. It consumes the parameters provided by the user and converts it to an instance of `BaseRelation` which is implemented by the class `com.databricks.spark.redshift.RedshiftRelation`. \n+\n+This `com.databricks.spark.redshift.RedshiftRelation` class is responsible for providing an `RDD` of `org.apache.spark.sql.Row` which backs the `org.apache.spark.sql.DataFrame` instance. This represents the underlying implementation for the Load functionality for the spark-redshift package where the schema is inferred from the underlying Redshift table. The functionality for performing load where the user provides the schema is supported by the trait `org.apache.spark.sql.sources.SchemaRelationProvider` and is implemented in the class `RedshiftRelation` . \n+\n+The store functionality of the spark-redshift package is supported by the trait `org.apache.spark.sql.sources.CreatableRelationProvider` implemented by the `RedshiftWriter`.\n+\n+\n+## Conclusion ###\n+\n+Data Sources API provides a unified interface for handling structured data in Spark. It was introduced in Spark 1.2 through the spark-sql package. It unifies structured data sources under a common interface which is richer than the original RDD based approach which utilized InputFormat API of Hadoop. In this article we explored how the spark-redshift package expands the Data Sources API to support Amazon Redshift which is becoming increasingly popular choice in the Enterprise IT infrastructure.","path":"tutorial/README.md","position":413,"original_position":413,"commit_id":"5a153937fdcb4700417051aeaa3d8a20e2be35cd","original_commit_id":"5a153937fdcb4700417051aeaa3d8a20e2be35cd","user":{"login":"JoshRosen","id":50748,"avatar_url":"https://avatars.githubusercontent.com/u/50748?v=3","gravatar_id":"","url":"https://api.github.com/users/JoshRosen","html_url":"https://github.com/JoshRosen","followers_url":"https://api.github.com/users/JoshRosen/followers","following_url":"https://api.github.com/users/JoshRosen/following{/other_user}","gists_url":"https://api.github.com/users/JoshRosen/gists{/gist_id}","starred_url":"https://api.github.com/users/JoshRosen/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/JoshRosen/subscriptions","organizations_url":"https://api.github.com/users/JoshRosen/orgs","repos_url":"https://api.github.com/users/JoshRosen/repos","events_url":"https://api.github.com/users/JoshRosen/events{/privacy}","received_events_url":"https://api.github.com/users/JoshRosen/received_events","type":"User","site_admin":false},"body":"_The_ data sources API? Spark's data sources API? With the latter, I'd drop the \"in Spark\" at the end of this sentence.\r\n\r\n> \"through the spark-sql package\"\r\n\r\nThis seems unnecessary and potentially-confusing.\r\n\r\n\"RDD based\" should be hyphenated.\r\n\r\n> \"utilized InputFormat API of Hadoop\"\r\n\r\nAgain, no \"utilized\". How about \"Hadoop's InputFormat API\"?","created_at":"2015-09-30T17:51:13Z","updated_at":"2015-09-30T17:51:13Z","html_url":"https://github.com/databricks/spark-redshift/pull/101#discussion_r40826790","pull_request_url":"https://api.github.com/repos/databricks/spark-redshift/pulls/101","_links":{"self":{"href":"https://api.github.com/repos/databricks/spark-redshift/pulls/comments/40826790"},"html":{"href":"https://github.com/databricks/spark-redshift/pull/101#discussion_r40826790"},"pull_request":{"href":"https://api.github.com/repos/databricks/spark-redshift/pulls/101"}}},"pull_request":{"url":"https://api.github.com/repos/databricks/spark-redshift/pulls/101","id":46403069,"html_url":"https://github.com/databricks/spark-redshift/pull/101","diff_url":"https://github.com/databricks/spark-redshift/pull/101.diff","patch_url":"https://github.com/databricks/spark-redshift/pull/101.patch","issue_url":"https://api.github.com/repos/databricks/spark-redshift/issues/101","number":101,"state":"open","locked":false,"title":"Tutorial","user":{"login":"sameeraxiomine","id":3198633,"avatar_url":"https://avatars.githubusercontent.com/u/3198633?v=3","gravatar_id":"","url":"https://api.github.com/users/sameeraxiomine","html_url":"https://github.com/sameeraxiomine","followers_url":"https://api.github.com/users/sameeraxiomine/followers","following_url":"https://api.github.com/users/sameeraxiomine/following{/other_user}","gists_url":"https://api.github.com/users/sameeraxiomine/gists{/gist_id}","starred_url":"https://api.github.com/users/sameeraxiomine/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/sameeraxiomine/subscriptions","organizations_url":"https://api.github.com/users/sameeraxiomine/orgs","repos_url":"https://api.github.com/users/sameeraxiomine/repos","events_url":"https://api.github.com/users/sameeraxiomine/events{/privacy}","received_events_url":"https://api.github.com/users/sameeraxiomine/received_events","type":"User","site_admin":false},"body":"Created the following set of files\r\n\r\n1. tutorial/README.md\r\n2. SparkRedshiftTutorial.scala\r\n3. images/loadreadstep.gif\r\n4. images/loadunloadstep.gif\r\n5. images/savetoredshift.gif\r\n\r\n","created_at":"2015-09-30T13:43:54Z","updated_at":"2015-09-30T17:51:13Z","closed_at":null,"merged_at":null,"merge_commit_sha":"4ac4f4e38dce937a58958dede5fe09db72f2831b","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/databricks/spark-redshift/pulls/101/commits","review_comments_url":"https://api.github.com/repos/databricks/spark-redshift/pulls/101/comments","review_comment_url":"https://api.github.com/repos/databricks/spark-redshift/pulls/comments{/number}","comments_url":"https://api.github.com/repos/databricks/spark-redshift/issues/101/comments","statuses_url":"https://api.github.com/repos/databricks/spark-redshift/statuses/5a153937fdcb4700417051aeaa3d8a20e2be35cd","head":{"label":"sameeraxiomine:master","ref":"master","sha":"5a153937fdcb4700417051aeaa3d8a20e2be35cd","user":{"login":"sameeraxiomine","id":3198633,"avatar_url":"https://avatars.githubusercontent.com/u/3198633?v=3","gravatar_id":"","url":"https://api.github.com/users/sameeraxiomine","html_url":"https://github.com/sameeraxiomine","followers_url":"https://api.github.com/users/sameeraxiomine/followers","following_url":"https://api.github.com/users/sameeraxiomine/following{/other_user}","gists_url":"https://api.github.com/users/sameeraxiomine/gists{/gist_id}","starred_url":"https://api.github.com/users/sameeraxiomine/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/sameeraxiomine/subscriptions","organizations_url":"https://api.github.com/users/sameeraxiomine/orgs","repos_url":"https://api.github.com/users/sameeraxiomine/repos","events_url":"https://api.github.com/users/sameeraxiomine/events{/privacy}","received_events_url":"https://api.github.com/users/sameeraxiomine/received_events","type":"User","site_admin":false},"repo":{"id":43395534,"name":"spark-redshift","full_name":"sameeraxiomine/spark-redshift","owner":{"login":"sameeraxiomine","id":3198633,"avatar_url":"https://avatars.githubusercontent.com/u/3198633?v=3","gravatar_id":"","url":"https://api.github.com/users/sameeraxiomine","html_url":"https://github.com/sameeraxiomine","followers_url":"https://api.github.com/users/sameeraxiomine/followers","following_url":"https://api.github.com/users/sameeraxiomine/following{/other_user}","gists_url":"https://api.github.com/users/sameeraxiomine/gists{/gist_id}","starred_url":"https://api.github.com/users/sameeraxiomine/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/sameeraxiomine/subscriptions","organizations_url":"https://api.github.com/users/sameeraxiomine/orgs","repos_url":"https://api.github.com/users/sameeraxiomine/repos","events_url":"https://api.github.com/users/sameeraxiomine/events{/privacy}","received_events_url":"https://api.github.com/users/sameeraxiomine/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/sameeraxiomine/spark-redshift","description":"Spark and Redshift integration","fork":true,"url":"https://api.github.com/repos/sameeraxiomine/spark-redshift","forks_url":"https://api.github.com/repos/sameeraxiomine/spark-redshift/forks","keys_url":"https://api.github.com/repos/sameeraxiomine/spark-redshift/keys{/key_id}","collaborators_url":"https://api.github.com/repos/sameeraxiomine/spark-redshift/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/sameeraxiomine/spark-redshift/teams","hooks_url":"https://api.github.com/repos/sameeraxiomine/spark-redshift/hooks","issue_events_url":"https://api.github.com/repos/sameeraxiomine/spark-redshift/issues/events{/number}","events_url":"https://api.github.com/repos/sameeraxiomine/spark-redshift/events","assignees_url":"https://api.github.com/repos/sameeraxiomine/spark-redshift/assignees{/user}","branches_url":"https://api.github.com/repos/sameeraxiomine/spark-redshift/branches{/branch}","tags_url":"https://api.github.com/repos/sameeraxiomine/spark-redshift/tags","blobs_url":"https://api.github.com/repos/sameeraxiomine/spark-redshift/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/sameeraxiomine/spark-redshift/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/sameeraxiomine/spark-redshift/git/refs{/sha}","trees_url":"https://api.github.com/repos/sameeraxiomine/spark-redshift/git/trees{/sha}","statuses_url":"https://api.github.com/repos/sameeraxiomine/spark-redshift/statuses/{sha}","languages_url":"https://api.github.com/repos/sameeraxiomine/spark-redshift/languages","stargazers_url":"https://api.github.com/repos/sameeraxiomine/spark-redshift/stargazers","contributors_url":"https://api.github.com/repos/sameeraxiomine/spark-redshift/contributors","subscribers_url":"https://api.github.com/repos/sameeraxiomine/spark-redshift/subscribers","subscription_url":"https://api.github.com/repos/sameeraxiomine/spark-redshift/subscription","commits_url":"https://api.github.com/repos/sameeraxiomine/spark-redshift/commits{/sha}","git_commits_url":"https://api.github.com/repos/sameeraxiomine/spark-redshift/git/commits{/sha}","comments_url":"https://api.github.com/repos/sameeraxiomine/spark-redshift/comments{/number}","issue_comment_url":"https://api.github.com/repos/sameeraxiomine/spark-redshift/issues/comments{/number}","contents_url":"https://api.github.com/repos/sameeraxiomine/spark-redshift/contents/{+path}","compare_url":"https://api.github.com/repos/sameeraxiomine/spark-redshift/compare/{base}...{head}","merges_url":"https://api.github.com/repos/sameeraxiomine/spark-redshift/merges","archive_url":"https://api.github.com/repos/sameeraxiomine/spark-redshift/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/sameeraxiomine/spark-redshift/downloads","issues_url":"https://api.github.com/repos/sameeraxiomine/spark-redshift/issues{/number}","pulls_url":"https://api.github.com/repos/sameeraxiomine/spark-redshift/pulls{/number}","milestones_url":"https://api.github.com/repos/sameeraxiomine/spark-redshift/milestones{/number}","notifications_url":"https://api.github.com/repos/sameeraxiomine/spark-redshift/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/sameeraxiomine/spark-redshift/labels{/name}","releases_url":"https://api.github.com/repos/sameeraxiomine/spark-redshift/releases{/id}","created_at":"2015-09-29T21:28:14Z","updated_at":"2015-09-29T21:28:15Z","pushed_at":"2015-09-30T13:39:09Z","git_url":"git://github.com/sameeraxiomine/spark-redshift.git","ssh_url":"git@github.com:sameeraxiomine/spark-redshift.git","clone_url":"https://github.com/sameeraxiomine/spark-redshift.git","svn_url":"https://github.com/sameeraxiomine/spark-redshift","homepage":null,"size":1789,"stargazers_count":0,"watchers_count":0,"language":"Scala","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"databricks:master","ref":"master","sha":"d1e77a038c95363545893f20a64b916e48f66c6c","user":{"login":"databricks","id":4998052,"avatar_url":"https://avatars.githubusercontent.com/u/4998052?v=3","gravatar_id":"","url":"https://api.github.com/users/databricks","html_url":"https://github.com/databricks","followers_url":"https://api.github.com/users/databricks/followers","following_url":"https://api.github.com/users/databricks/following{/other_user}","gists_url":"https://api.github.com/users/databricks/gists{/gist_id}","starred_url":"https://api.github.com/users/databricks/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/databricks/subscriptions","organizations_url":"https://api.github.com/users/databricks/orgs","repos_url":"https://api.github.com/users/databricks/repos","events_url":"https://api.github.com/users/databricks/events{/privacy}","received_events_url":"https://api.github.com/users/databricks/received_events","type":"Organization","site_admin":false},"repo":{"id":26562411,"name":"spark-redshift","full_name":"databricks/spark-redshift","owner":{"login":"databricks","id":4998052,"avatar_url":"https://avatars.githubusercontent.com/u/4998052?v=3","gravatar_id":"","url":"https://api.github.com/users/databricks","html_url":"https://github.com/databricks","followers_url":"https://api.github.com/users/databricks/followers","following_url":"https://api.github.com/users/databricks/following{/other_user}","gists_url":"https://api.github.com/users/databricks/gists{/gist_id}","starred_url":"https://api.github.com/users/databricks/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/databricks/subscriptions","organizations_url":"https://api.github.com/users/databricks/orgs","repos_url":"https://api.github.com/users/databricks/repos","events_url":"https://api.github.com/users/databricks/events{/privacy}","received_events_url":"https://api.github.com/users/databricks/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/databricks/spark-redshift","description":"Spark and Redshift integration","fork":false,"url":"https://api.github.com/repos/databricks/spark-redshift","forks_url":"https://api.github.com/repos/databricks/spark-redshift/forks","keys_url":"https://api.github.com/repos/databricks/spark-redshift/keys{/key_id}","collaborators_url":"https://api.github.com/repos/databricks/spark-redshift/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/databricks/spark-redshift/teams","hooks_url":"https://api.github.com/repos/databricks/spark-redshift/hooks","issue_events_url":"https://api.github.com/repos/databricks/spark-redshift/issues/events{/number}","events_url":"https://api.github.com/repos/databricks/spark-redshift/events","assignees_url":"https://api.github.com/repos/databricks/spark-redshift/assignees{/user}","branches_url":"https://api.github.com/repos/databricks/spark-redshift/branches{/branch}","tags_url":"https://api.github.com/repos/databricks/spark-redshift/tags","blobs_url":"https://api.github.com/repos/databricks/spark-redshift/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/databricks/spark-redshift/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/databricks/spark-redshift/git/refs{/sha}","trees_url":"https://api.github.com/repos/databricks/spark-redshift/git/trees{/sha}","statuses_url":"https://api.github.com/repos/databricks/spark-redshift/statuses/{sha}","languages_url":"https://api.github.com/repos/databricks/spark-redshift/languages","stargazers_url":"https://api.github.com/repos/databricks/spark-redshift/stargazers","contributors_url":"https://api.github.com/repos/databricks/spark-redshift/contributors","subscribers_url":"https://api.github.com/repos/databricks/spark-redshift/subscribers","subscription_url":"https://api.github.com/repos/databricks/spark-redshift/subscription","commits_url":"https://api.github.com/repos/databricks/spark-redshift/commits{/sha}","git_commits_url":"https://api.github.com/repos/databricks/spark-redshift/git/commits{/sha}","comments_url":"https://api.github.com/repos/databricks/spark-redshift/comments{/number}","issue_comment_url":"https://api.github.com/repos/databricks/spark-redshift/issues/comments{/number}","contents_url":"https://api.github.com/repos/databricks/spark-redshift/contents/{+path}","compare_url":"https://api.github.com/repos/databricks/spark-redshift/compare/{base}...{head}","merges_url":"https://api.github.com/repos/databricks/spark-redshift/merges","archive_url":"https://api.github.com/repos/databricks/spark-redshift/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/databricks/spark-redshift/downloads","issues_url":"https://api.github.com/repos/databricks/spark-redshift/issues{/number}","pulls_url":"https://api.github.com/repos/databricks/spark-redshift/pulls{/number}","milestones_url":"https://api.github.com/repos/databricks/spark-redshift/milestones{/number}","notifications_url":"https://api.github.com/repos/databricks/spark-redshift/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/databricks/spark-redshift/labels{/name}","releases_url":"https://api.github.com/repos/databricks/spark-redshift/releases{/id}","created_at":"2014-11-13T00:08:13Z","updated_at":"2015-09-30T14:27:08Z","pushed_at":"2015-09-30T13:43:54Z","git_url":"git://github.com/databricks/spark-redshift.git","ssh_url":"git@github.com:databricks/spark-redshift.git","clone_url":"https://github.com/databricks/spark-redshift.git","svn_url":"https://github.com/databricks/spark-redshift","homepage":null,"size":1789,"stargazers_count":87,"watchers_count":87,"language":"Scala","has_issues":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":34,"mirror_url":null,"open_issues_count":18,"forks":34,"open_issues":18,"watchers":87,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/databricks/spark-redshift/pulls/101"},"html":{"href":"https://github.com/databricks/spark-redshift/pull/101"},"issue":{"href":"https://api.github.com/repos/databricks/spark-redshift/issues/101"},"comments":{"href":"https://api.github.com/repos/databricks/spark-redshift/issues/101/comments"},"review_comments":{"href":"https://api.github.com/repos/databricks/spark-redshift/pulls/101/comments"},"review_comment":{"href":"https://api.github.com/repos/databricks/spark-redshift/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/databricks/spark-redshift/pulls/101/commits"},"statuses":{"href":"https://api.github.com/repos/databricks/spark-redshift/statuses/5a153937fdcb4700417051aeaa3d8a20e2be35cd"}}}}