diff --git a/docs/streaming-kafka-0-10-integration.md b/docs/streaming-kafka-0-10-integration.md
index c1ef396907..b645d3c3a4 100644
--- a/docs/streaming-kafka-0-10-integration.md
+++ b/docs/streaming-kafka-0-10-integration.md
@@ -17,69 +17,72 @@ For Scala/Java applications using SBT/Maven project definitions, link your strea
- 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
+{% highlight scala %}
+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
- val kafkaParams = Map[String, Object](
- "bootstrap.servers" -> "localhost:9092,anotherhost:9092",
- "key.deserializer" -> classOf[StringDeserializer],
- "value.deserializer" -> classOf[StringDeserializer],
- "group.id" -> "use_a_separate_group_id_for_each_stream",
- "auto.offset.reset" -> "latest",
- "enable.auto.commit" -> (false: java.lang.Boolean)
- )
+val kafkaParams = Map[String, Object](
+ "bootstrap.servers" -> "localhost:9092,anotherhost:9092",
+ "key.deserializer" -> classOf[StringDeserializer],
+ "value.deserializer" -> classOf[StringDeserializer],
+ "group.id" -> "use_a_separate_group_id_for_each_stream",
+ "auto.offset.reset" -> "latest",
+ "enable.auto.commit" -> (false: java.lang.Boolean)
+)
- val topics = Array("topicA", "topicB")
- val stream = KafkaUtils.createDirectStream[String, String](
- streamingContext,
- PreferConsistent,
- Subscribe[String, String](topics, kafkaParams)
- )
-
- stream.map(record => (record.key, record.value))
+val topics = Array("topicA", "topicB")
+val stream = KafkaUtils.createDirectStream[String, String](
+ streamingContext,
+ PreferConsistent,
+ Subscribe[String, String](topics, kafkaParams)
+)
+stream.map(record => (record.key, record.value))
+{% endhighlight %}
Each item in the stream is a [ConsumerRecord](http://kafka.apache.org/0100/javadoc/org/apache/kafka/clients/consumer/ConsumerRecord.html)
- import java.util.*;
- import org.apache.spark.SparkConf;
- import org.apache.spark.TaskContext;
- import org.apache.spark.api.java.*;
- import org.apache.spark.api.java.function.*;
- import org.apache.spark.streaming.api.java.*;
- import org.apache.spark.streaming.kafka010.*;
- import org.apache.kafka.clients.consumer.ConsumerRecord;
- import org.apache.kafka.common.TopicPartition;
- import org.apache.kafka.common.serialization.StringDeserializer;
- import scala.Tuple2;
-
- Map kafkaParams = new HashMap<>();
- kafkaParams.put("bootstrap.servers", "localhost:9092,anotherhost:9092");
- kafkaParams.put("key.deserializer", StringDeserializer.class);
- kafkaParams.put("value.deserializer", StringDeserializer.class);
- kafkaParams.put("group.id", "use_a_separate_group_id_for_each_stream");
- kafkaParams.put("auto.offset.reset", "latest");
- kafkaParams.put("enable.auto.commit", false);
-
- Collection topics = Arrays.asList("topicA", "topicB");
-
- final JavaInputDStream> stream =
- KafkaUtils.createDirectStream(
- streamingContext,
- LocationStrategies.PreferConsistent(),
- ConsumerStrategies.Subscribe(topics, kafkaParams)
- );
-
- stream.mapToPair(
- new PairFunction, String, String>() {
- @Override
- public Tuple2 call(ConsumerRecord record) {
- return new Tuple2<>(record.key(), record.value());
- }
- })
+{% highlight java %}
+import java.util.*;
+import org.apache.spark.SparkConf;
+import org.apache.spark.TaskContext;
+import org.apache.spark.api.java.*;
+import org.apache.spark.api.java.function.*;
+import org.apache.spark.streaming.api.java.*;
+import org.apache.spark.streaming.kafka010.*;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import scala.Tuple2;
+
+Map kafkaParams = new HashMap<>();
+kafkaParams.put("bootstrap.servers", "localhost:9092,anotherhost:9092");
+kafkaParams.put("key.deserializer", StringDeserializer.class);
+kafkaParams.put("value.deserializer", StringDeserializer.class);
+kafkaParams.put("group.id", "use_a_separate_group_id_for_each_stream");
+kafkaParams.put("auto.offset.reset", "latest");
+kafkaParams.put("enable.auto.commit", false);
+
+Collection topics = Arrays.asList("topicA", "topicB");
+
+final JavaInputDStream> stream =
+ KafkaUtils.createDirectStream(
+ streamingContext,
+ LocationStrategies.PreferConsistent(),
+ ConsumerStrategies.Subscribe(topics, kafkaParams)
+ );
+
+stream.mapToPair(
+ new PairFunction, String, String>() {
+ @Override
+ public Tuple2 call(ConsumerRecord record) {
+ return new Tuple2<>(record.key(), record.value());
+ }
+ })
+{% endhighlight %}
@@ -109,32 +112,35 @@ If you have a use case that is better suited to batch processing, you can create
- // Import dependencies and create kafka params as in Create Direct Stream above
+{% highlight scala %}
+// Import dependencies and create kafka params as in Create Direct Stream above
- val offsetRanges = Array(
- // topic, partition, inclusive starting offset, exclusive ending offset
- OffsetRange("test", 0, 0, 100),
- OffsetRange("test", 1, 0, 100)
- )
-
- val rdd = KafkaUtils.createRDD[String, String](sparkContext, kafkaParams, offsetRanges, PreferConsistent)
+val offsetRanges = Array(
+ // topic, partition, inclusive starting offset, exclusive ending offset
+ OffsetRange("test", 0, 0, 100),
+ OffsetRange("test", 1, 0, 100)
+)
+val rdd = KafkaUtils.createRDD[String, String](sparkContext, kafkaParams, offsetRanges, PreferConsistent)
+{% endhighlight %}
- // Import dependencies and create kafka params as in Create Direct Stream above
+{% highlight java %}
+// Import dependencies and create kafka params as in Create Direct Stream above
- OffsetRange[] offsetRanges = {
- // topic, partition, inclusive starting offset, exclusive ending offset
- OffsetRange.create("test", 0, 0, 100),
- OffsetRange.create("test", 1, 0, 100)
- };
+OffsetRange[] offsetRanges = {
+ // topic, partition, inclusive starting offset, exclusive ending offset
+ OffsetRange.create("test", 0, 0, 100),
+ OffsetRange.create("test", 1, 0, 100)
+};
- JavaRDD> rdd = KafkaUtils.createRDD(
- sparkContext,
- kafkaParams,
- offsetRanges,
- LocationStrategies.PreferConsistent()
- );
+JavaRDD> rdd = KafkaUtils.createRDD(
+ sparkContext,
+ kafkaParams,
+ offsetRanges,
+ LocationStrategies.PreferConsistent()
+);
+{% endhighlight %}
@@ -144,29 +150,33 @@ Note that you cannot use `PreferBrokers`, because without the stream there is no
- stream.foreachRDD { rdd =>
- 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}")
- }
- }
+{% highlight scala %}
+stream.foreachRDD { rdd =>
+ 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}")
+ }
+}
+{% endhighlight %}
- stream.foreachRDD(new VoidFunction>>() {
- @Override
- public void call(JavaRDD> rdd) {
- final OffsetRange[] offsetRanges = ((HasOffsetRanges) rdd.rdd()).offsetRanges();
- rdd.foreachPartition(new VoidFunction>>() {
- @Override
- public void call(Iterator> consumerRecords) {
- OffsetRange o = offsetRanges[TaskContext.get().partitionId()];
- System.out.println(
- o.topic() + " " + o.partition() + " " + o.fromOffset() + " " + o.untilOffset());
- }
- });
- }
- });
+{% highlight java %}
+stream.foreachRDD(new VoidFunction>>() {
+ @Override
+ public void call(JavaRDD> rdd) {
+ final OffsetRange[] offsetRanges = ((HasOffsetRanges) rdd.rdd()).offsetRanges();
+ rdd.foreachPartition(new VoidFunction>>() {
+ @Override
+ public void call(Iterator> consumerRecords) {
+ OffsetRange o = offsetRanges[TaskContext.get().partitionId()];
+ System.out.println(
+ o.topic() + " " + o.partition() + " " + o.fromOffset() + " " + o.untilOffset());
+ }
+ });
+ }
+});
+{% endhighlight %}
@@ -183,25 +193,28 @@ Kafka has an offset commit API that stores offsets in a special Kafka topic. By
- stream.foreachRDD { rdd =>
- val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
-
- // some time later, after outputs have completed
- stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
- }
+{% highlight scala %}
+stream.foreachRDD { rdd =>
+ val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
+ // some time later, after outputs have completed
+ stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
+}
+{% endhighlight %}
As with HasOffsetRanges, the cast to CanCommitOffsets will only succeed if called on the result of createDirectStream, not after transformations. The commitAsync call is threadsafe, but must occur after outputs if you want meaningful semantics.
- stream.foreachRDD(new VoidFunction>>() {
- @Override
- public void call(JavaRDD> rdd) {
- OffsetRange[] offsetRanges = ((HasOffsetRanges) rdd.rdd()).offsetRanges();
+{% highlight java %}
+stream.foreachRDD(new VoidFunction>>() {
+ @Override
+ public void call(JavaRDD> rdd) {
+ OffsetRange[] offsetRanges = ((HasOffsetRanges) rdd.rdd()).offsetRanges();
- // some time later, after outputs have completed
- ((CanCommitOffsets) stream.inputDStream()).commitAsync(offsetRanges);
- }
- });
+ // some time later, after outputs have completed
+ ((CanCommitOffsets) stream.inputDStream()).commitAsync(offsetRanges);
+ }
+});
+{% endhighlight %}
@@ -210,64 +223,68 @@ For data stores that support transactions, saving offsets in the same transactio
- // The details depend on your data store, but the general idea looks like this
+{% highlight scala %}
+// The details depend on your data store, but the general idea looks like this
- // begin from the the offsets committed to the database
- val fromOffsets = selectOffsetsFromYourDatabase.map { resultSet =>
- new TopicPartition(resultSet.string("topic"), resultSet.int("partition")) -> resultSet.long("offset")
- }.toMap
+// begin from the the offsets committed to the database
+val fromOffsets = selectOffsetsFromYourDatabase.map { resultSet =>
+ new TopicPartition(resultSet.string("topic"), resultSet.int("partition")) -> resultSet.long("offset")
+}.toMap
- val stream = KafkaUtils.createDirectStream[String, String](
- streamingContext,
- PreferConsistent,
- Assign[String, String](fromOffsets.keys.toList, kafkaParams, fromOffsets)
- )
+val stream = KafkaUtils.createDirectStream[String, String](
+ streamingContext,
+ PreferConsistent,
+ Assign[String, String](fromOffsets.keys.toList, kafkaParams, fromOffsets)
+)
- stream.foreachRDD { rdd =>
- val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
+stream.foreachRDD { rdd =>
+ val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
- val results = yourCalculation(rdd)
+ val results = yourCalculation(rdd)
- // begin your transaction
+ // begin your transaction
- // update results
- // update offsets where the end of existing offsets matches the beginning of this batch of offsets
- // assert that offsets were updated correctly
+ // update results
+ // update offsets where the end of existing offsets matches the beginning of this batch of offsets
+ // assert that offsets were updated correctly
- // end your transaction
- }
+ // end your transaction
+}
+{% endhighlight %}
- // The details depend on your data store, but the general idea looks like this
+{% highlight java %}
+// The details depend on your data store, but the general idea looks like this
- // begin from the the offsets committed to the database
- Map fromOffsets = new HashMap<>();
- for (resultSet : selectOffsetsFromYourDatabase)
- fromOffsets.put(new TopicPartition(resultSet.string("topic"), resultSet.int("partition")), resultSet.long("offset"));
- }
+// begin from the the offsets committed to the database
+Map fromOffsets = new HashMap<>();
+for (resultSet : selectOffsetsFromYourDatabase)
+ fromOffsets.put(new TopicPartition(resultSet.string("topic"), resultSet.int("partition")), resultSet.long("offset"));
+}
- JavaInputDStream> stream = KafkaUtils.createDirectStream(
- streamingContext,
- LocationStrategies.PreferConsistent(),
- ConsumerStrategies.Assign(fromOffsets.keySet(), kafkaParams, fromOffsets)
- );
+JavaInputDStream> stream = KafkaUtils.createDirectStream(
+ streamingContext,
+ LocationStrategies.PreferConsistent(),
+ ConsumerStrategies.Assign(fromOffsets.keySet(), kafkaParams, fromOffsets)
+);
- stream.foreachRDD(new VoidFunction>>() {
- @Override
- public void call(JavaRDD> rdd) {
- OffsetRange[] offsetRanges = ((HasOffsetRanges) rdd.rdd()).offsetRanges();
-
- Object results = yourCalculation(rdd);
+stream.foreachRDD(new VoidFunction>>() {
+ @Override
+ public void call(JavaRDD> rdd) {
+ OffsetRange[] offsetRanges = ((HasOffsetRanges) rdd.rdd()).offsetRanges();
+
+ Object results = yourCalculation(rdd);
- // begin your transaction
+ // begin your transaction
- // update results
- // update offsets where the end of existing offsets matches the beginning of this batch of offsets
- // assert that offsets were updated correctly
+ // update results
+ // update offsets where the end of existing offsets matches the beginning of this batch of offsets
+ // assert that offsets were updated correctly
- // end your transaction
- }
- });
+ // end your transaction
+ }
+});
+{% endhighlight %}
@@ -277,25 +294,29 @@ The new Kafka consumer [supports SSL](http://kafka.apache.org/documentation.html
- val kafkaParams = Map[String, Object](
- // the usual params, make sure to change the port in bootstrap.servers if 9092 is not TLS
- "security.protocol" -> "SSL",
- "ssl.truststore.location" -> "/some-directory/kafka.client.truststore.jks",
- "ssl.truststore.password" -> "test1234",
- "ssl.keystore.location" -> "/some-directory/kafka.client.keystore.jks",
- "ssl.keystore.password" -> "test1234",
- "ssl.key.password" -> "test1234"
- )
+{% highlight scala %}
+val kafkaParams = Map[String, Object](
+ // the usual params, make sure to change the port in bootstrap.servers if 9092 is not TLS
+ "security.protocol" -> "SSL",
+ "ssl.truststore.location" -> "/some-directory/kafka.client.truststore.jks",
+ "ssl.truststore.password" -> "test1234",
+ "ssl.keystore.location" -> "/some-directory/kafka.client.keystore.jks",
+ "ssl.keystore.password" -> "test1234",
+ "ssl.key.password" -> "test1234"
+)
+{% endhighlight %}
- Map kafkaParams = new HashMap();
- // the usual params, make sure to change the port in bootstrap.servers if 9092 is not TLS
- kafkaParams.put("security.protocol", "SSL");
- kafkaParams.put("ssl.truststore.location", "/some-directory/kafka.client.truststore.jks");
- kafkaParams.put("ssl.truststore.password", "test1234");
- kafkaParams.put("ssl.keystore.location", "/some-directory/kafka.client.keystore.jks");
- kafkaParams.put("ssl.keystore.password", "test1234");
- kafkaParams.put("ssl.key.password", "test1234");
+{% highlight java %}
+Map kafkaParams = new HashMap();
+// the usual params, make sure to change the port in bootstrap.servers if 9092 is not TLS
+kafkaParams.put("security.protocol", "SSL");
+kafkaParams.put("ssl.truststore.location", "/some-directory/kafka.client.truststore.jks");
+kafkaParams.put("ssl.truststore.password", "test1234");
+kafkaParams.put("ssl.keystore.location", "/some-directory/kafka.client.keystore.jks");
+kafkaParams.put("ssl.keystore.password", "test1234");
+kafkaParams.put("ssl.key.password", "test1234");
+{% endhighlight %}
diff --git a/docs/structured-streaming-kafka-integration.md b/docs/structured-streaming-kafka-integration.md
index a6c3b3a902..c4c9fb3f7d 100644
--- a/docs/structured-streaming-kafka-integration.md
+++ b/docs/structured-streaming-kafka-integration.md
@@ -19,97 +19,103 @@ application. See the [Deploying](#deploying) subsection below.
+{% highlight scala %}
- // Subscribe to 1 topic
- val ds1 = spark
- .readStream
- .format("kafka")
- .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
- .option("subscribe", "topic1")
- .load()
- ds1.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
- .as[(String, String)]
+// Subscribe to 1 topic
+val ds1 = spark
+ .readStream
+ .format("kafka")
+ .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
+ .option("subscribe", "topic1")
+ .load()
+ds1.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
+ .as[(String, String)]
- // Subscribe to multiple topics
- val ds2 = spark
- .readStream
- .format("kafka")
- .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
- .option("subscribe", "topic1,topic2")
- .load()
- ds2.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
- .as[(String, String)]
+// Subscribe to multiple topics
+val ds2 = spark
+ .readStream
+ .format("kafka")
+ .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
+ .option("subscribe", "topic1,topic2")
+ .load()
+ds2.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
+ .as[(String, String)]
- // Subscribe to a pattern
- val ds3 = spark
- .readStream
- .format("kafka")
- .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
- .option("subscribePattern", "topic.*")
- .load()
- ds3.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
- .as[(String, String)]
+// Subscribe to a pattern
+val ds3 = spark
+ .readStream
+ .format("kafka")
+ .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
+ .option("subscribePattern", "topic.*")
+ .load()
+ds3.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
+ .as[(String, String)]
+{% endhighlight %}
+{% highlight java %}
- // Subscribe to 1 topic
- Dataset ds1 = spark
- .readStream()
- .format("kafka")
- .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
- .option("subscribe", "topic1")
- .load()
- ds1.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
+// Subscribe to 1 topic
+Dataset ds1 = spark
+ .readStream()
+ .format("kafka")
+ .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
+ .option("subscribe", "topic1")
+ .load()
+ds1.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
- // Subscribe to multiple topics
- Dataset ds2 = spark
- .readStream()
- .format("kafka")
- .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
- .option("subscribe", "topic1,topic2")
- .load()
- ds2.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
+// Subscribe to multiple topics
+Dataset ds2 = spark
+ .readStream()
+ .format("kafka")
+ .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
+ .option("subscribe", "topic1,topic2")
+ .load()
+ds2.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
- // Subscribe to a pattern
- Dataset ds3 = spark
- .readStream()
- .format("kafka")
- .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
- .option("subscribePattern", "topic.*")
- .load()
- ds3.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
+// Subscribe to a pattern
+Dataset ds3 = spark
+ .readStream()
+ .format("kafka")
+ .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
+ .option("subscribePattern", "topic.*")
+ .load()
+ds3.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
+{% endhighlight %}
+{% highlight python %}
- # Subscribe to 1 topic
- ds1 = spark
- .readStream()
- .format("kafka")
- .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
- .option("subscribe", "topic1")
- .load()
- ds1.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
+# Subscribe to 1 topic
+ds1 = spark
+ .readStream()
+ .format("kafka")
+ .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
+ .option("subscribe", "topic1")
+ .load()
+ds1.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
- # Subscribe to multiple topics
- ds2 = spark
- .readStream
- .format("kafka")
- .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
- .option("subscribe", "topic1,topic2")
- .load()
- ds2.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
+# Subscribe to multiple topics
+ds2 = spark
+ .readStream
+ .format("kafka")
+ .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
+ .option("subscribe", "topic1,topic2")
+ .load()
+ds2.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
- # Subscribe to a pattern
- ds3 = spark
- .readStream()
- .format("kafka")
- .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
- .option("subscribePattern", "topic.*")
- .load()
- ds3.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
+# Subscribe to a pattern
+ds3 = spark
+ .readStream()
+ .format("kafka")
+ .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
+ .option("subscribePattern", "topic.*")
+ .load()
+ds3.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
+{% endhighlight %}