Add example about how to convert key and value to strings

Author: zsxwing

docs/structured-streaming-kafka-integration.md

@@ -22,82 +22,94 @@ application. See the [Deploying](#deploying) subsection below.
 <div data-lang="scala" markdown="1">
 
     // Subscribe to 1 topic
-    spark
+    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
-    spark
+    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
-    spark
+    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)]
 
 </div>
 <div data-lang="java" markdown="1">
 
     // Subscribe to 1 topic
-    spark
+    Dataset<Row> 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
-    spark
+    Dataset<Row> 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
-    spark
+    Dataset<Row> 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)")
 
 </div>
 <div data-lang="python" markdown="1">
 
     # Subscribe to 1 topic
-    spark
+    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
-    spark
+    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
-    spark
+    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)")
 
 </div>
 </div>

external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaSourceSuite.scala

@@ -145,8 +145,10 @@ class KafkaSourceSuite extends KafkaSourceTest {
       .option("kafka.metadata.max.age.ms", "1")
       .option("subscribePattern", s"topic-.*")
 
-    val kafka = reader.load().select("key", "value").as[(Array[Byte], Array[Byte])]
-    val mapped = kafka.map(kv => new String(kv._2).toInt + 1)
+    val kafka = reader.load()
+      .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
+      .as[(String, String)]
+    val mapped = kafka.map(kv => kv._2.toInt + 1)
 
     testStream(mapped)(
       StopStream
@@ -190,8 +192,10 @@ class KafkaSourceSuite extends KafkaSourceTest {
       .option("kafka.metadata.max.age.ms", "1")
       .option("subscribePattern", s"$topicPrefix-.*")
 
-    val kafka = reader.load().select("key", "value").as[(Array[Byte], Array[Byte])]
-    val mapped = kafka.map(kv => new String(kv._2).toInt + 1)
+    val kafka = reader.load()
+      .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
+      .as[(String, String)]
+    val mapped = kafka.map(kv => kv._2.toInt + 1)
 
     testStream(mapped)(
       makeSureGetOffsetCalled,
@@ -272,8 +276,10 @@ class KafkaSourceSuite extends KafkaSourceTest {
       .option("kafka.bootstrap.servers", testUtils.brokerAddress)
       .option("kafka.metadata.max.age.ms", "1")
     options.foreach { case (k, v) => reader.option(k, v) }
-    val kafka = reader.load().select("key", "value").as[(Array[Byte], Array[Byte])]
-    val mapped = kafka.map(kv => new String(kv._2).toInt + 1)
+    val kafka = reader.load()
+      .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
+      .as[(String, String)]
+    val mapped = kafka.map(kv => kv._2.toInt + 1)
 
     testStream(mapped)(
       makeSureGetOffsetCalled,
@@ -309,8 +315,10 @@ class KafkaSourceSuite extends KafkaSourceTest {
       .option("kafka.bootstrap.servers", testUtils.brokerAddress)
       .option("kafka.metadata.max.age.ms", "1")
     options.foreach { case (k, v) => reader.option(k, v) }
-    val kafka = reader.load().select("key", "value").as[(Array[Byte], Array[Byte])]
-    val mapped = kafka.map(kv => new String(kv._2).toInt + 1)
+    val kafka = reader.load()
+      .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
+      .as[(String, String)]
+    val mapped = kafka.map(kv => kv._2.toInt + 1)
 
     testStream(mapped)(
       AddKafkaData(Set(topic), 4, 5, 6), // Add data when stream is stopped
@@ -368,10 +376,10 @@ class KafkaSourceStressSuite extends KafkaSourceTest with BeforeAndAfter {
         .option("subscribePattern", "stress.*")
         .option("failOnCorruptMetadata", "false")
         .load()
-        .select("key", "value")
-        .as[(Array[Byte], Array[Byte])]
+        .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
+        .as[(String, String)]
 
-    val mapped = kafka.map(kv => new String(kv._2).toInt + 1)
+    val mapped = kafka.map(kv => kv._2.toInt + 1)
 
     runStressTest(
       mapped,