load kafka config from application.conf (KafkaEventLoad/LoadKafkaEven…

…tExample.scala
yennanliu · May 31, 2020 · b2aa1b3 · b2aa1b3
1 parent 0cababc
commit b2aa1b3
Showing 1 changed file with 10 additions and 0 deletions.
diff --git a/src/main/scala/KafkaEventLoad/LoadKafkaEventExample.scala b/src/main/scala/KafkaEventLoad/LoadKafkaEventExample.scala
@@ -16,6 +16,8 @@ import org.apache.spark.sql.streaming.{OutputMode, Trigger}
 import org.apache.spark.sql.types.{StringType, StructField, StructType}
 import org.apache.spark.sql.functions.{get_json_object, json_tuple}
 
+import com.typesafe.config.ConfigFactory
+
 /*
  * modify from 
  * https://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html
@@ -26,6 +28,8 @@ object LoadKafkaEventExample {
 
   def main(args: Array[String]): Unit = {
 
+      // spark config 
+
       val sc = new SparkContext("local[*]", "LoadKafkaEventExample")   
       val sqlContext = new org.apache.spark.sql.SQLContext(sc)
       val spark = SparkSession
@@ -42,6 +46,12 @@ object LoadKafkaEventExample {
 
       import spark.implicits._
 
+      // kafka config
+
+      val kafkaconfig = ConfigFactory.load().getConfig("kafka")
+      val bootstrapservers:String = kafkaconfig.getString("BOOTSTRAP_SERVERS")
+      val topic:String = kafkaconfig.getString("TOPIC")
+
       // Subscribe to 1 topic
 
       val df = spark