在scala spark Streaming中使用foreach时不希望string作为类型？

Question

代码段：

val lines = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap)
val write2hdfs = lines.filter(x => x._1 == "lineitem").map(_._2)
write2hdfs.foreachRDD(rdd => {

rdd.foreach(avroRecord => {
println(avroRecord)
//val rawByte = avroRecord.getBytes("UTF-8")

val ssc = new StreamingContext(sparkConf, Seconds(5))
//val ssc = new JavaStreamingContext(sparkConf, Seconds(5))
val topicMap = topics.split(",").map((_, numThreads.toInt)).toMap
val kafkaParams = Map[String, String]("zookeeper.connect" -> 
zkQuorum,"group.id" -> group,"zookeeper.connection.timeout.ms" -> "10000")                    

//val lines = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap)
val lines = 
KafkaUtils.createStream[String,Message,StringDecoder,DefaultDecoder]
(ssc,kafkaParams,topics,StorageLevel.NONE)

已完成的进口：

import org.apache.spark.streaming._
import org.apache.spark.streaming.api.java.JavaStreamingContext
import org.apache.spark.streaming.kafka._
import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.DStream.toPairDStreamFunctions
import org.apache.avro
import org.apache.avro.Schema
import org.apache.avro.generic.{GenericDatumReader, GenericRecord, 
GenericDatumWriter, GenericData}
import org.apache.avro.io.{DecoderFactory, DatumReader, DatumWriter, 
BinaryDecoder}
import org.apache.avro.file.{DataFileReader, DataFileWriter}
import java.io.{File, IOException}
//import java.io.*
import org.apache.commons.io.IOUtils;
import _root_.kafka.serializer.{StringDecoder, DefaultDecoder}
import _root_.kafka.message.Message
import scala.reflect._

编译错误：

将1个Scala源代码编译为/home/spark_scala/spark_stream_project/target/scala-2.10/classes...[error]/home/spark_scala/spark_stream_project/src/main/scala/sparkstreaming.scala:34:重载方法值createStream，其备选方案Elevel:org.apache.spark.storage.storageLevel)org.apache.spark.streaming.api.java.javapairReceiverInputdStream[String,kafka.message.message][error](Ssc:org.apache.spark.streaming.streaming.streamingContext,kafkaparams:scala.collection.immutable.map[String，String]，topics:scala.collection.immutable.map[String，String]，stopics:scala.collection.immutable.map[String，String]，stopics:scala.collection,age.message],隐式证据$3:scala.reflect.classtag[kafka.serializer.StringDecoder],隐式证据$4:scala.reflect.classtag[kafka.serializer.DefaultDecoder])org.apache.spark.streaming.dreceiverInputdStream[(String,kafka.Message.Message)][(String,kafka.Message.Message)]不能应用于（org.apache.spark.streamingcontext,scala.collection.immutable.map[String,String,org.apache.spark.streamingcontext,String,org.apache.spark.storagelevel)[error]

这里出了什么问题。另外，我没有看到kafkaUtils API文档中所建议的正确构造函数。API文档引用：https://spark.apache.org/docs/1.3.0/API/java/index.html？org/apache/spark/streaming/kafkautils.html

期待支持。

谢了。

val lines = 
KafkaUtils.createStream[String,Message,StringDecoder,DefaultDecoder]
(ssc,kafkaParams,topicMap,StorageLevel.MEMORY_AND_DISK_2)
val write2hdfs = lines.filter(x => x._1 == "lineitem").map(_._2)

java.lang.ClassCastException: [B cannot be cast to kafka.message.Message

On line :
KafkaUtils.createStream[String,Message,StringDecoder,DefaultDecoder]
(ssc,kafkaParams,topicMap,StorageLevel.MEMORY_AND_DISK_2)
val write2hdfs = lines.filter(x => x._1 == "lineitem").map(_._2)

理想情况下，过滤这个Dstream（字符串、消息）应该也可以工作，对吗？在使用map之前，我需要从消息中提取有效负载吗？

需要输入请。谢谢

Answer 1

您可以这样做：

import kafka.serializer.{StringDecoder, DefaultDecoder}
import kafka.message.Message

val kafkaParams = Map[String, String](
    "zookeeper.connect" -> zkQuorum, "group.id" -> group,
    "zookeeper.connection.timeout.ms" -> "10000")
val lines = KafkaUtils.createStream[String, Message, StringDecoder, DefaultDecoder](
      ssc, kafkaParams, topics, storageLevel)

这将得到DStream[(String，kafka.message.message)]，您应该能够检索原始字节并从那里转换为Avro。