Apache Kafka and Apache Spark (simplest connection for newbies)
Useful and needed links:
1) download and unpack Spark and Kafka
https://spark.apache.org/downloads.html
tar xvzf spark-xxxx.tgz
https://kafka.apache.org/downloads
tar xvzf kafka-xxxx.tgz
2) Run Kafka (zookeeper and broker in different consoles):
console 1:
cd kafka-xxxx
bin/zookeeper-server-start.sh config/zookeeper.properties
console 2:
cd kafka-xxxx
bin/kafka-server-start.sh config/server.properties
console 3
- create new topic:
bin/kafka-topics.sh --create --topic angel --bootstrap-server localhost:9092
- chech that topic is created and available:
bin/kafka-topics.sh --list --bootstrap-server localhost:9092
or
bin/kafka-topics.sh --describe --topic angel --bootstrap-server localhost:9092
Check that it works with Kafka console producer/consumer (in different consoles):
-producer:
bin/kafka-console-producer.sh --topic angel --bootstrap-server localhost:9092
- consumer:
bin/kafka-console-consumer.sh --topic angel --bootstrap-server localhost:9092
- in producer type in some symbols to see that they will apear in console consumer
3) Download needed libraries to Kafka jars libraries:
cd spark-3.0.1-bin-hadoop2.7/jars
wget jars from links:
https://mvnrepository.com/artifact/or...
https://mvnrepository.com/artifact/or...
https://mvnrepository.com/artifact/or...
https://mvnrepository.com/artifact/or...
https://mvnrepository.com/artifact/or...
4) Run Spark-shell with reading messages from Kafka:
cd ../bin
./spark-shell --jars ../jars/kafka-clients-2.6.0.jar, ../jars/spark-streaming-kafka-0-10_2.12-3.0.1.jar, ../jars/spark-sql-kafka-0-10_2.12-3.0.1.jar, ../jars/spark-streaming-kafka-0-10-assembly_2.12-3.0.1.jar ../jars/commons-pool2-2.9.0.jar
In spark-shell to import neede packages/classes type in:
import org.apache.spark._
import org.apache.spark.streaming._
import org.apache.spark.streaming.StreamingContext._
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import spark.implicits._
In spark-shell to import neede packages/classes type in:
val df = spark.readStream.format("kafka").option("kafka.bootstrap.servers", "localhost:9092").option("subscribe","andriytopic").load()
- to see the schema:
df.printSchema
- to read messages:
val query = df.writeStream.outputMode("append").format("console").start().awaitTermination()