Help & Documentation>TDMQ for CKafka

Connecting Spark Streaming to CKafka

Last updated: 2026-01-05 16:55:12

On this page:

As an extension of Spark Core, Spark Streaming is used for high-throughput and fault-tolerant processing of continuous data. Currently supported external input sources include Kafka, Flume, HDFS/S3, Kinesis, Twitter, and TCP socket.



Spark Streaming abstracts continuous data into DStream (Discretized Stream), which is composed of a series of continuous RDDs (Resilient Distributed Datasets). Each RDD represents data generated over a certain time interval. Processing a DStream using functions is essentially processing these RDDs.



When Spark Streaming is used as data input for Kafka, the following stable and experimental Kafka versions are supported:
Kafka Version
spark-streaming-kafka-0.8
spark-streaming-kafka-0.10
Broker Version
0.8.2.1 or higher
0.10.0 or higher
Api Maturity
Deprecated
Stable
Language Support
Scala、Java、Python
Scala、Java
Receiver DStream
Yes
No
Direct DStream
Yes
Yes
SSL / TLS Support
No
Yes
Offset Commit Api
No
Yes
Dynamic Topic Subscription
No
Yes
Currently, CKafka is compatible with version above 0.9. The Kafka dependency of v0.10.2.1 is used in this practice scenario.
Additionally, Spark Streaming in EMR also supports direct integration with CKafka. For more information, see Spark Streaming Integration with CKafka Service.

Instructions

Step 1. Get the CKafka instance access address

1. Log in to the CKafka console.
2. In the left navigation bar, select Instance List, then click on the instance "ID" to access the basic information page of the instance.
3. On the Access Mode module of the instance's basic information page, you can obtain the instance's access address, which is the bootstrap-server required for production and consumption.



Step 2. Create a topic

1. On the instance basic information page, select the Topic Management tab at the top.
2. On the Topic Management page, click Create to establish a Topic named 'test'. The following discussion will use this Topic as an example to illustrate how to produce and consume.



Step 3. Prepare the CVM environment

CentOS 6.8 System
package
version
sbt
0.13.16
hadoop
2.7.3
spark
2.1.0
protobuf
2.5.0
ssh
Default CentOS Installation
Java
1.8
For specific installation steps, refer to Environment Configuration.

Step 4. Connect to CKafka

Produce messages to CKafka
Consuming Messages from CKafka
The Kafka dependency of v0.10.2.1 is used here.
1. Add dependencies to the build.sbt file:
name := "Producer Example"
version := "1.0"
scalaVersion := "2.11.8"
libraryDependencies += "org.apache.kafka" % "kafka-clients" % "0.10.2.1"
2. Configure producer_example.scala:
   import java.util.Properties
   import org.apache.kafka.clients.producer._
   object ProducerExample extends App {
    val  props = new Properties()
    props.put("bootstrap.servers", "172.16.16.12:9092") //Private IP and port from the instance information

    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")

    val producer = new KafkaProducer[String, String](props)
    val TOPIC="test"  //Specify the Topic to produce
    for(i<- 1 to 50){
           val record = new ProducerRecord(TOPIC, "key", s"hello $i") //Produces a message with the key as "key" and value as "hello i"
           producer.send(record)
    }
    val record = new ProducerRecord(TOPIC, "key", "the end "+new java.util.Date)
    producer.send(record)
    producer.close() // Finally, disconnect
   }
For more information on the usage of ProducerRecord, please see the ProducerRecord documentation.

DirectStream

1. Add dependencies to the build.sbt file:
name := "Consumer Example"
version := "1.0"
scalaVersion := "2.11.8"
libraryDependencies += "org.apache.spark" %% "spark-core" % "2.1.0"
libraryDependencies += "org.apache.spark" %% "spark-streaming" % "2.1.0"
libraryDependencies += "org.apache.spark" %% "spark-streaming-kafka-0-10" % "2.1.0"
2. Configure DirectStream_example.scala:
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.kafka.common.TopicPartition
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.OffsetRange
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import collection.JavaConversions._
import Array._
object Kafka {
    def main(args: Array[String]) {
        val kafkaParams = Map[String, Object](
            "bootstrap.servers" -> "172.16.16.12:9092",
            "key.deserializer" -> classOf[StringDeserializer],
            "value.deserializer" -> classOf[StringDeserializer],
            "group.id" -> "spark_stream_test1",
            "auto.offset.reset" -> "earliest",
            "enable.auto.commit" -> "false"
        )

        val sparkConf = new SparkConf()
        sparkConf.setMaster("local")
        sparkConf.setAppName("Kafka")
        val ssc = new StreamingContext(sparkConf, Seconds(5))
        val topics = Array("spark_test")

        val offsets : Map[TopicPartition, Long] = Map()

        for (i <- 0 until 3){
            val tp = new TopicPartition("spark_test", i)
            offsets.updated(tp , 0L)
        }
        val stream = KafkaUtils.createDirectStream[String, String](
            ssc,
            PreferConsistent,
            Subscribe[String, String](topics, kafkaParams)
        )
        println("directStream")
        stream.foreachRDD{ rdd=>
            // Output the received message
            rdd.foreach{iter =>
                val i = iter.value
                println(s"${i}")
            }
            //Acquiring the offset
            val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
            rdd.foreachPartition { iter =>
                val o: OffsetRange = offsetRanges(TaskContext.get.partitionId)
                println(s"${o.topic} ${o.partition} ${o.fromOffset} ${o.untilOffset}")
            }
        }

        // Start the computation
        ssc.start()
        ssc.awaitTermination()
    }
}

RDD

1. Configure build.sbt (configuration is the same as above, click to view).
2. Configuration of RDD_example:
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.OffsetRange
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import collection.JavaConversions._
import Array._
object Kafka {
    def main(args: Array[String]) {
        val kafkaParams = Map[String, Object](
            "bootstrap.servers" -> "172.16.16.12:9092",
            "key.deserializer" -> classOf[StringDeserializer],
            "value.deserializer" -> classOf[StringDeserializer],
            "group.id" -> "spark_stream",
            "auto.offset.reset" -> "earliest",
            "enable.auto.commit" -> (false: java.lang.Boolean)
        )
        val sc = new SparkContext("local", "Kafka", new SparkConf())
        val java_kafkaParams : java.util.Map[String, Object] = kafkaParams
        //Fetch messages from the partition in the specified offset range sequentially. If no messages are available, block until the wait time is exceeded or the required number of new messages is produced.
        val offsetRanges = Array[OffsetRange](
            OffsetRange("spark_test", 0, 0, 5),
            OffsetRange("spark_test", 1, 0, 5),
            OffsetRange("spark_test", 2, 0, 5)
        )
        val range = KafkaUtils.createRDD[String, String](
            sc,
            java_kafkaParams,
            offsetRanges,
            PreferConsistent
        )
        range.foreach(rdd=>println(rdd.value))
        sc.stop()
    }
}
For more information on the usage of kafkaParams, refer to the kafkaParams documentation.

Environment Configuration

Installing sbt

1. Download the sbt package from the sbt official website.
2. After decompression, create an sbt_run.sh script with the following content in the sbt directory and add executable permissions:
#!/bin/bash
SBT_OPTS="-Xms512M -Xmx1536M -Xss1M -XX:+CMSClassUnloadingEnabled -XX:MaxPermSize=256M"
java $SBT_OPTS -jar dirname $0/bin/sbt-launch.jar "$@"
chmod u+x ./sbt_run.sh
3. Run the following command:
./sbt-run.sh sbt-version
The display of sbt version indicates a successful installation.

Installing Protobuf

1. Download the corresponding version of protobuf.
2. Decompress and enter the directory.
./configure
make && make install
You should install gcc-g++ in advance, and the root permission may be required during installation.
3. Log in again and enter the following on the command line:
protoc --version
4. The display of Protobuf version indicates a successful installation.

Installing Hadoop

1. Visit the Hadoop official website to download the desired version.
2. Add a Hadoop user.
useradd -m hadoop -s /bin/bash
3. Grant admin permissions.
visudo
4. Add a new line below root ALL=(ALL) ALL: hadoop ALL=(ALL) ALL. Save and exit.
5. Use Hadoop for operations.
su hadoop
6. Configure SSH password-free login.
cd ~/.ssh/                     # If there is no such directory, run ssh localhost first
ssh-keygen -t rsa              # There will be prompts. Simply press Enter
cat id_rsa.pub >> authorized_keys  # Add authorization
chmod 600 ./authorized_keys    # Modify file permission
7. Install Java.
sudo yum install java-1.8.0-openjdk java-1.8.0-openjdk-devel
8. Configure ${JAVA_HOME}.
vim /etc/profile
Add the following at the end:
export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.121-0.b13.el6_8.x86_64/jre
export PATH=$PATH:$JAVA_HOME
Modify the corresponding path based on the installation information.
9. Decompress Hadoop and enter the directory.
./bin/hadoop version
The display of version information indicates a successful installation.
10. Configure a pseudo-distributed single-node setup (different types of clusters can be built as per requirements).
vim /etc/profile
Add the following at the end:
export HADOOP_HOME=/usr/local/hadoop
export PATH=$HADOOP_HOME/bin:$PATH
Modify the corresponding path based on the installation information.
11. Modify the /etc/hadoop/core-site.xml file.
<configuration>    
    <property>       
        <name>hadoop.tmp.dir</name>       
        <value>file:/usr/local/hadoop/tmp</value>
        <description>Abase for other temporary directories.</description>
    </property>    
    <property>        
        <name>fs.defaultFS</name>
        <value>hdfs://localhost:9000</value> 
    </property>
</configuration>
12. Modify the /etc/hadoop/hdfs-site.xml file.
<configuration>
    <property>
        <name>dfs.replication</name>
        <value>1</value>
    </property>
    <property>
        <name>dfs.namenode.name.dir</name>
        <value>file:/usr/local/hadoop/tmp/dfs/name</value>
    </property>
    <property>
        <name>dfs.datanode.data.dir</name>
        <value>file:/usr/local/hadoop/tmp/dfs/data</value>
    </property>
</configuration>
13. Modify the JAVA_HOME in /etc/hadoop/hadoop-env.sh to the path of Java.
export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.121-0.b13.el6_8.x86_64/jre
14. Perform NameNode formatting.
./bin/hdfs namenode -format
If Exitting with status 0 is displayed, the operation is successful.
15. Initiate Hadoop.
./sbin/start-dfs.sh
Upon successful startup, the following processes will be present: NameNode, DataNode, and SecondaryNameNode.

Installing Spark

Visit the Spark official website to download the desired version. Since Hadoop has been previously installed, select Pre-built with user-provided Apache Hadoop.
Note
In this example, the hadoop user is also used for the operations.
1. Decompress and enter the directory.
2. Modify the configuration file.
cp ./conf/spark-env.sh.template ./conf/spark-env.sh
vim ./conf/spark-env.sh
Add the following in the first line:
export SPARK_DIST_CLASSPATH=$(/usr/local/hadoop/bin/hadoop classpath)
Modify the path based on the Hadoop installation information.
3. Run the example.
bin/run-example SparkPi
The display of an approximate value of π output by the program indicates a successful installation.
中国站 - English