title: Provider org.apache.spark.sql.avro.AvroFileFormat could not be instantiated
body: <p>Unable to send avro format message to Kafka topic from spark streaming application. Very less information is available online about avro spark streaming example code. "to_avro" method doesn't require avro schema then how it will encode to avro format?</p>
  
  <p><strong>Can someone please help to resolve below exception?</strong></p>
  
  <p>Dependency:</p>
  
  <pre><code>&lt;dependency&gt;
      &lt;groupId&gt;org.apache.spark&lt;/groupId&gt;
      &lt;artifactId&gt;spark-avro_2.12&lt;/artifactId&gt;
      &lt;version&gt;2.4.4&lt;/version&gt;
  &lt;/dependency&gt;
  &lt;dependency&gt;
      &lt;groupId&gt;org.apache.spark&lt;/groupId&gt;
      &lt;artifactId&gt;spark-core_2.11&lt;/artifactId&gt;
      &lt;version&gt;2.4.0&lt;/version&gt;
  &lt;/dependency&gt;
  &lt;dependency&gt;
      &lt;groupId&gt;org.apache.spark&lt;/groupId&gt;
      &lt;artifactId&gt;spark-streaming-kafka-0-10_2.11&lt;/artifactId&gt;
      &lt;version&gt;2.4.0&lt;/version&gt;
  &lt;/dependency&gt;
  </code></pre>
  
  <p>Below is the code to push to kafka topic</p>
  
  <pre><code>dataset.toDF.select(to_avro(struct(dataset.toDF.columns.map(column):_*))).alias("value").distinct.write.format("avro")
        .option(KafkaConstants.BOOTSTRAP_SERVER, priBootStrapServers)
        .option(ApplicationConstants.TOPIC_KEY, publishPriTopic)
        .save()
  </code></pre>
  
  <p>Getting below exception.</p>
  
  <pre><code>Caused by: java.util.ServiceConfigurationError: org.apache.spark.sql.sources.DataSourceRegister: Provider org.apache.spark.sql.avro.AvroFileFormat could not be instantiated
      at java.util.ServiceLoader.fail(ServiceLoader.java:232)
      at java.util.ServiceLoader.access$100(ServiceLoader.java:185)
      at java.util.ServiceLoader$LazyIterator.nextService(ServiceLoader.java:384)
      at java.util.ServiceLoader$LazyIterator.next(ServiceLoader.java:404)
      at java.util.ServiceLoader$1.next(ServiceLoader.java:480)
      at scala.collection.convert.Wrappers$JIteratorWrapper.next(Wrappers.scala:43)
      at scala.collection.Iterator$class.foreach(Iterator.scala:893)
      at scala.collection.AbstractIterator.foreach(Iterator.scala:1336)
      at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
      at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
      at scala.collection.TraversableLike$class.filterImpl(TraversableLike.scala:247)
      at scala.collection.TraversableLike$class.filter(TraversableLike.scala:259)
      at scala.collection.AbstractTraversable.filter(Traversable.scala:104)
      at org.apache.spark.sql.execution.datasources.DataSource$.lookupDataSource(DataSource.scala:614)
      at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:241)
      at com.walmart.replenishment.edf.dao.EdfOwBuzzerDao$.saveToCassandra(EdfOwBuzzerDao.scala:47)
      at com.walmart.replenishment.edf.process.BuzzerService$.updateScrItemPriStatus(BuzzerService.scala:119)
      at com.walmart.replenishment.edf.process.BuzzerStreamProcessor$$anonfun$processConsumerInputStream$1.apply(BuzzerStreamProcessor.scala:36)
      at com.walmart.replenishment.edf.process.BuzzerStreamProcessor$$anonfun$processConsumerInputStream$1.apply(BuzzerStreamProcessor.scala:28)
      at org.apache.spark.streaming.dstream.DStream$$anonfun$foreachRDD$1$$anonfun$apply$mcV$sp$3.apply(DStream.scala:628)
      at org.apache.spark.streaming.dstream.DStream$$anonfun$foreachRDD$1$$anonfun$apply$mcV$sp$3.apply(DStream.scala:628)
      at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(ForEachDStream.scala:51)
      at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1$$anonfun$apply$mcV$sp$1.apply(ForEachDStream.scala:51)
      at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1$$anonfun$apply$mcV$sp$1.apply(ForEachDStream.scala:51)
      at org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:416)
      at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply$mcV$sp(ForEachDStream.scala:50)
      at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:50)
      at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:50)
      at scala.util.Try$.apply(Try.scala:192)
      at org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)
      at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply$mcV$sp(JobScheduler.scala:257)
      at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply(JobScheduler.scala:257)
      at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply(JobScheduler.scala:257)
      at scala.util.DynamicVariable.withValue(DynamicVariable.scala:58)
      at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
      at java.lang.Thread.run(Thread.java:745)
  Caused by: java.lang.NoSuchMethodError: org.apache.spark.sql.execution.datasources.FileFormat.$init$(Lorg/apache/spark/sql/execution/datasources/FileFormat;)V
      at org.apache.spark.sql.avro.AvroFileFormat.(AvroFileFormat.scala:44)
      at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
      at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
      at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
      at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
      at java.lang.Class.newInstance(Class.java:442)
      at java.util.ServiceLoader$LazyIterator.nextService(ServiceLoader.java:380)
      ```
  
  
  </code></pre>