title: how to save apache spark schema output in mysql database
body: <p>Can anyone please tell me if there is any way in apache spark to store a JavaRDD on mysql database? I am taking input from 2 csv files and then after doing join operations on their contents I need to save the output(the output JavaRDD) in the mysql database. I am already able to save the output successfully on hdfs but I am not finding any information related to apache Spark-MYSQL connection. Below I am posting the code for spark sql. This might serve as a reference to those who are looking for an example for spark-sql.</p>
  
  <pre><code>package attempt1;
  
  import java.io.Serializable;
  
  import org.apache.spark.api.java.JavaRDD;
  import org.apache.spark.api.java.JavaSparkContext;
  import org.apache.spark.api.java.function.Function;
  import org.apache.spark.sql.api.java.JavaSQLContext;
  import org.apache.spark.sql.api.java.JavaSchemaRDD;
  import org.apache.spark.sql.api.java.Row;
  
  
  public class Spark_Mysql {
      @SuppressWarnings("serial")
      public static class CompleteSample implements Serializable {
          private String ASSETNUM;
          private String ASSETTAG;
          private String CALNUM;
  
  
  
          public String getASSETNUM() {
              return ASSETNUM;
          }
          public void setASSETNUM(String aSSETNUM) {
              ASSETNUM = aSSETNUM;
          }
          public String getASSETTAG() {
              return ASSETTAG;
          }
          public void setASSETTAG(String aSSETTAG) {
              ASSETTAG = aSSETTAG;
          }
          public String getCALNUM() {
              return CALNUM;
          }
          public void setCALNUM(String cALNUM) {
              CALNUM = cALNUM;
          }
  
  
        }
  
      @SuppressWarnings("serial")
      public static class ExtendedSample implements Serializable {
  
          private String ASSETNUM;
          private String CHANGEBY;
          private String CHANGEDATE;
  
  
          public String getASSETNUM() {
              return ASSETNUM;
          }
          public void setASSETNUM(String aSSETNUM) {
              ASSETNUM = aSSETNUM;
          }
          public String getCHANGEBY() {
              return CHANGEBY;
          }
          public void setCHANGEBY(String cHANGEBY) {
              CHANGEBY = cHANGEBY;
          }
          public String getCHANGEDATE() {
              return CHANGEDATE;
          }
          public void setCHANGEDATE(String cHANGEDATE) {
              CHANGEDATE = cHANGEDATE;
          }
      }
  
      @SuppressWarnings("serial")
      public static void main(String[] args) throws Exception {
  
            JavaSparkContext ctx = new JavaSparkContext("local[2]", "JavaSparkSQL");
            JavaSQLContext sqlCtx = new JavaSQLContext(ctx);
  
            JavaRDD&lt;CompleteSample&gt; cs = ctx.textFile("C:/Users/cyg_server/Documents/bigDataExample/AssetsImportCompleteSample.csv").map(
                    new Function&lt;String, CompleteSample&gt;() {
                      public CompleteSample call(String line) throws Exception {
                        String[] parts = line.split(",");
  
                        CompleteSample cs = new CompleteSample();
                        cs.setASSETNUM(parts[0]);
                        cs.setASSETTAG(parts[1]);
                        cs.setCALNUM(parts[2]);
  
                        return cs;
                      }
                    });
  
            JavaRDD&lt;ExtendedSample&gt; es = ctx.textFile("C:/Users/cyg_server/Documents/bigDataExample/AssetsImportExtendedSample.csv").map(
                    new Function&lt;String, ExtendedSample&gt;() {
                      public ExtendedSample call(String line) throws Exception {
                        String[] parts = line.split(",");
  
                        ExtendedSample es = new ExtendedSample();
                        es.setASSETNUM(parts[0]);
                        es.setCHANGEBY(parts[1]);
                        es.setCHANGEDATE(parts[2]);
  
                        return es;
                      }
                    });
  
            JavaSchemaRDD complete = sqlCtx.applySchema(cs, CompleteSample.class);
              complete.registerAsTable("cs");
  
            JavaSchemaRDD extended = sqlCtx.applySchema(es, ExtendedSample.class);
            extended.registerAsTable("es");
  
            JavaSchemaRDD fs= sqlCtx.sql("SELECT cs.ASSETTAG, cs.CALNUM, es.CHANGEBY, es.CHANGEDATE FROM cs INNER JOIN es ON cs.ASSETNUM=es.ASSETNUM;");
  
  
            JavaRDD&lt;String&gt; result = fs.map(new Function&lt;Row, String&gt;() {
                public String call(Row row) {
                  return row.getString(0);
                }
              });
  
                result.saveAsTextFile("hdfs://path/to/hdfs/dir-name");          //instead of hdfs I need to save it on mysql database, but I am not able to find any Spark-MYSQL connection
  
      }
  
  
  
  }
  </code></pre>
  
  <p>Here at the end I am saving the result successfully in HDFS. But now I want to save into MYSQL database. Kindly help me out. Thanks</p>