title: Spark Cassandra connector write Dataframe to Cassandra
body: <p>I have a dataframe which I want to write to Cassandra table. The data frame consists of 4,680,820 rows. The data itself is coming from .csv file which is around 650MB large and read by spark Cassandra connector from datastax. </p>
  
  <p>Then, using Spark Cassandra connector to write this data in dataframe to cassandra. The connector does write records but not all. It is currently only writing 17349 records. How should I optimize the writes and also write all the 4.6 million records. I have 8 Exeuctors and each executor has 4 cores each so today of 28 tasks can be run in parallel. </p>
  
  <p>I am using Cassandra 3.0.13 and Spark 2.1.0 and Spark Cassandra Connector:spark-cassandra-connector-assembly-2.0.1</p>
  
  <p>This is the code I am using: </p>
  
  <pre><code> def writeToCassandra(df: DataFrame, tableName: String) {
      if (df.take(1).length&gt;0){
          println("Number of rows in DF are: "+ df.count);
          df.write
          .format("org.apache.spark.sql.cassandra")
          .mode("overwrite")
          .options(Map(
              "header" -&gt; "true",
              "output.batch.grouping.key" -&gt; "none",
              "cassandra.output.throughput_mb_per_sec" -&gt; "1MB",
              "table" -&gt; tableName, 
              "keyspace" -&gt; "vardb_irc", 
              "cluster" -&gt; "CassandraPOCluster")
              )
          .save()
      }
      else{
          println("Dataframe is empty");
      }
  }
  </code></pre>
  
  <p>The above method is called by below method.</p>
  
  <pre><code>def runCSVDataSourceImport(){
      // Load the CM data into with SPARkSQL
      val adErDF = spark.read
        .format("com.databricks.spark.csv")
        .option("header", "true")
        .option("inferSchema", "true")
        .load("file:///opt/risk-4.20170511.csv")
  
     println("DF: adErDF ready to use")
     println(adErDF.count);
     val adErDFRepartition = adErDF.repartition(3)
     adErDFRepartition.explain
     writeToCassandra(adErDFRepartition, "risk_4");
  }
  </code></pre>
  
  <p>thanks</p>