The problem is "optional Gender gender = 3;". The generated class "Gender"
is a trait, and Spark cannot know how to create a trait so it's not
supported. You can define your class which is supported by SQL Encoder, and
convert this generated class to the new class in `parseLine`.
On Wed, Nov 16, 2016 at 4:22 PM, shyla deshpande <deshpandesh...@gmail.com>
wrote:
> Ryan,
>
> I just wanted to provide more info. Here is my .proto file which is the
> basis for generating the Person class. Thanks.
>
> option java_package = "com.example.protos";
> enum Gender {
> MALE = 1;
> FEMALE = 2;
> }
> message Address {
> optional string street = 1;
> optional string city = 2;
> }
> message Person {
> optional string name = 1;
> optional int32 age = 2;
> optional Gender gender = 3;
> repeated string tags = 4;
> repeated Address addresses = 5;
> }
>
>
> On Wed, Nov 16, 2016 at 3:04 PM, shyla deshpande <deshpandesh...@gmail.com
> > wrote:
>
>> *Thanks for the response. Following is the Person class..*
>>
>> // Generated by the Scala Plugin for the Protocol Buffer Compiler.
>> // Do not edit!
>> //
>> // Protofile syntax: PROTO2
>>
>> package com.example.protos.demo
>>
>>
>>
>> @SerialVersionUID(0L)
>> final case class Person(
>> name: scala.Option[String] = None,
>> age: scala.Option[Int] = None,
>> gender: scala.Option[com.example.protos.demo.Gender] = None,
>> tags: scala.collection.Seq[String] = Nil,
>> addresses: scala.collection.Seq[com.example.protos.demo.Address] = Nil
>> ) extends com.trueaccord.scalapb.GeneratedMessage with
>> com.trueaccord.scalapb.Message[Person] with
>> com.trueaccord.lenses.Updatable[Person] {
>> @transient
>> private[this] var __serializedSizeCachedValue: Int = 0
>> private[this] def __computeSerializedValue(): Int = {
>> var __size = 0
>> if (name.isDefined) { __size +=
>> com.google.protobuf.CodedOutputStream.computeStringSize(1, name.get) }
>> if (age.isDefined) { __size +=
>> com.google.protobuf.CodedOutputStream.computeInt32Size(2, age.get) }
>> if (gender.isDefined) { __size +=
>> com.google.protobuf.CodedOutputStream.computeEnumSize(3, gender.get.value) }
>> tags.foreach(tags => __size +=
>> com.google.protobuf.CodedOutputStream.computeStringSize(4, tags))
>> addresses.foreach(addresses => __size += 1 +
>> com.google.protobuf.CodedOutputStream.computeUInt32SizeNoTag(addresses.serializedSize)
>> + addresses.serializedSize)
>> __size
>> }
>> final override def serializedSize: Int = {
>> var read = __serializedSizeCachedValue
>> if (read == 0) {
>> read = __computeSerializedValue()
>> __serializedSizeCachedValue = read
>> }
>> read
>> }
>> def writeTo(`_output__`: com.google.protobuf.CodedOutputStream): Unit = {
>> name.foreach { __v =>
>> _output__.writeString(1, __v)
>> };
>> age.foreach { __v =>
>> _output__.writeInt32(2, __v)
>> };
>> gender.foreach { __v =>
>> _output__.writeEnum(3, __v.value)
>> };
>> tags.foreach { __v =>
>> _output__.writeString(4, __v)
>> };
>> addresses.foreach { __v =>
>> _output__.writeTag(5, 2)
>> _output__.writeUInt32NoTag(__v.serializedSize)
>> __v.writeTo(_output__)
>> };
>> }
>> def mergeFrom(`_input__`: com.google.protobuf.CodedInputStream):
>> com.example.protos.demo.Person = {
>> var __name = this.name
>> var __age = this.age
>> var __gender = this.gender
>> val __tags = (scala.collection.immutable.Vector.newBuilder[String] ++=
>> this.tags)
>> val __addresses =
>> (scala.collection.immutable.Vector.newBuilder[com.example.protos.demo.Address]
>> ++= this.addresses)
>> var _done__ = false
>> while (!_done__) {
>> val _tag__ = _input__.readTag()
>> _tag__ match {
>> case 0 => _done__ = true
>> case 10 =>
>> __name = Some(_input__.readString())
>> case 16 =>
>> __age = Some(_input__.readInt32())
>> case 24 =>
>> __gender =
>> Some(com.example.protos.demo.Gender.fromValue(_input__.readEnum()))
>> case 34 =>
>> __tags += _input__.readString()
>> case 42 =>
>> __addresses +=
>> com.trueaccord.scalapb.LiteParser.readMessage(_input__,
>> com.example.protos.demo.Address.defaultInstance)
>> case tag => _input__.skipField(tag)
>> }
>> }
>> com.example.protos.demo.Person(
>> name = __name,
>> age = __age,
>> gender = __gender,
>> tags = __tags.result(),
>> addresses = __addresses.result()
>> )
>> }
>> def getName: String = name.getOrElse("")
>> def clearName: Person = copy(name = None)
>> def withName(__v: String): Person = copy(name = Some(__v))
>> def getAge: Int = age.getOrElse(0)
>> def clearAge: Person = copy(age = None)
>> def withAge(__v: Int): Person = copy(age = Some(__v))
>> def getGender: com.example.protos.demo.Gender =
>> gender.getOrElse(com.example.protos.demo.Gender.MALE)
>> def clearGender: Person = copy(gender = None)
>> def withGender(__v: com.example.protos.demo.Gender): Person =
>> copy(gender = Some(__v))
>> def clearTags = copy(tags = scala.collection.Seq.empty)
>> def addTags(__vs: String*): Person = addAllTags(__vs)
>> def addAllTags(__vs: TraversableOnce[String]): Person = copy(tags = tags
>> ++ __vs)
>> def withTags(__v: scala.collection.Seq[String]): Person = copy(tags =
>> __v)
>> def clearAddresses = copy(addresses = scala.collection.Seq.empty)
>> def addAddresses(__vs: com.example.protos.demo.Address*): Person =
>> addAllAddresses(__vs)
>> def addAllAddresses(__vs:
>> TraversableOnce[com.example.protos.demo.Address]): Person = copy(addresses =
>> addresses ++ __vs)
>> def withAddresses(__v:
>> scala.collection.Seq[com.example.protos.demo.Address]): Person =
>> copy(addresses = __v)
>> def getField(__field: com.google.protobuf.Descriptors.FieldDescriptor):
>> scala.Any = {
>> __field.getNumber match {
>> case 1 => name.getOrElse(null)
>> case 2 => age.getOrElse(null)
>> case 3 => gender.map(_.valueDescriptor).getOrElse(null)
>> case 4 => tags
>> case 5 => addresses
>> }
>> }
>> override def toString: String =
>> com.trueaccord.scalapb.TextFormat.printToUnicodeString(this)
>> def companion = com.example.protos.demo.Person
>> }
>>
>> object Person extends
>> com.trueaccord.scalapb.GeneratedMessageCompanion[com.example.protos.demo.Person]
>> {
>> implicit def messageCompanion:
>> com.trueaccord.scalapb.GeneratedMessageCompanion[com.example.protos.demo.Person]
>> = this
>> def fromFieldsMap(__fieldsMap:
>> scala.collection.immutable.Map[com.google.protobuf.Descriptors.FieldDescriptor,
>> scala.Any]): com.example.protos.demo.Person = {
>> require(__fieldsMap.keys.forall(_.getContainingType() == descriptor),
>> "FieldDescriptor does not match message type.")
>> val __fields = descriptor.getFields
>> com.example.protos.demo.Person(
>> __fieldsMap.get(__fields.get(0)).asInstanceOf[scala.Option[String]],
>> __fieldsMap.get(__fields.get(1)).asInstanceOf[scala.Option[Int]],
>>
>> __fieldsMap.get(__fields.get(2)).asInstanceOf[scala.Option[com.google.protobuf.Descriptors.EnumValueDescriptor]].map(__e
>> => com.example.protos.demo.Gender.fromValue(__e.getNumber)),
>> __fieldsMap.getOrElse(__fields.get(3),
>> Nil).asInstanceOf[scala.collection.Seq[String]],
>> __fieldsMap.getOrElse(__fields.get(4),
>> Nil).asInstanceOf[scala.collection.Seq[com.example.protos.demo.Address]]
>> )
>> }
>> def descriptor: com.google.protobuf.Descriptors.Descriptor =
>> DemoProto.descriptor.getMessageTypes.get(1)
>> def messageCompanionForField(__field:
>> com.google.protobuf.Descriptors.FieldDescriptor):
>> com.trueaccord.scalapb.GeneratedMessageCompanion[_] = {
>> require(__field.getContainingType() == descriptor, "FieldDescriptor does
>> not match message type.")
>> var __out: com.trueaccord.scalapb.GeneratedMessageCompanion[_] = null
>> __field.getNumber match {
>> case 5 => __out = com.example.protos.demo.Address
>> }
>> __out
>> }
>> def enumCompanionForField(__field:
>> com.google.protobuf.Descriptors.FieldDescriptor):
>> com.trueaccord.scalapb.GeneratedEnumCompanion[_] = {
>> require(__field.getContainingType() == descriptor, "FieldDescriptor does
>> not match message type.")
>> __field.getNumber match {
>> case 3 => com.example.protos.demo.Gender
>> }
>> }
>> lazy val defaultInstance = com.example.protos.demo.Person(
>> )
>> implicit class PersonLens[UpperPB](_l: com.trueaccord.lenses.Lens[UpperPB,
>> com.example.protos.demo.Person]) extends
>> com.trueaccord.lenses.ObjectLens[UpperPB,
>> com.example.protos.demo.Person](_l) {
>> def name: com.trueaccord.lenses.Lens[UpperPB, String] =
>> field(_.getName)((c_, f_) => c_.copy(name = Some(f_)))
>> def optionalName: com.trueaccord.lenses.Lens[UpperPB,
>> scala.Option[String]] = field(_.name)((c_, f_) => c_.copy(name = f_))
>> def age: com.trueaccord.lenses.Lens[UpperPB, Int] = field(_.getAge)((c_,
>> f_) => c_.copy(age = Some(f_)))
>> def optionalAge: com.trueaccord.lenses.Lens[UpperPB, scala.Option[Int]]
>> = field(_.age)((c_, f_) => c_.copy(age = f_))
>> def gender: com.trueaccord.lenses.Lens[UpperPB,
>> com.example.protos.demo.Gender] = field(_.getGender)((c_, f_) =>
>> c_.copy(gender = Some(f_)))
>> def optionalGender: com.trueaccord.lenses.Lens[UpperPB,
>> scala.Option[com.example.protos.demo.Gender]] = field(_.gender)((c_, f_) =>
>> c_.copy(gender = f_))
>> def tags: com.trueaccord.lenses.Lens[UpperPB,
>> scala.collection.Seq[String]] = field(_.tags)((c_, f_) => c_.copy(tags = f_))
>> def addresses: com.trueaccord.lenses.Lens[UpperPB,
>> scala.collection.Seq[com.example.protos.demo.Address]] =
>> field(_.addresses)((c_, f_) => c_.copy(addresses = f_))
>> }
>> final val NAME_FIELD_NUMBER = 1
>> final val AGE_FIELD_NUMBER = 2
>> final val GENDER_FIELD_NUMBER = 3
>> final val TAGS_FIELD_NUMBER = 4
>> final val ADDRESSES_FIELD_NUMBER = 5
>> }
>>
>>
>> On Wed, Nov 16, 2016 at 1:28 PM, Shixiong(Ryan) Zhu <
>> shixi...@databricks.com> wrote:
>>
>>> Could you provide the Person class?
>>>
>>> On Wed, Nov 16, 2016 at 1:19 PM, shyla deshpande <
>>> deshpandesh...@gmail.com> wrote:
>>>
>>>> I am using 2.11.8. Thanks
>>>>
>>>> On Wed, Nov 16, 2016 at 1:15 PM, Shixiong(Ryan) Zhu <
>>>> shixi...@databricks.com> wrote:
>>>>
>>>>> Which Scala version are you using? Is it Scala 2.10? Scala 2.10 has
>>>>> some known race conditions in reflection and the Scala community doesn't
>>>>> have plan to fix it (http://docs.scala-lang.org/ov
>>>>> erviews/reflection/thread-safety.html) AFAIK, the only way to fix it
>>>>> is upgrading to Scala 2.11.
>>>>>
>>>>> On Wed, Nov 16, 2016 at 11:16 AM, shyla deshpande <
>>>>> deshpandesh...@gmail.com> wrote:
>>>>>
>>>>>> I am using protobuf to encode. This may not be related to the new
>>>>>> release issue....
>>>>>>
>>>>>> Exception in thread "main" scala.ScalaReflectionException: <none> is
>>>>>> not a term
>>>>>> at scala.reflect.api.Symbols$SymbolApi$class.asTerm(Symbols.sca
>>>>>> la:199)
>>>>>> at scala.reflect.internal.Symbols$SymbolContextApiImpl.asTerm(S
>>>>>> ymbols.scala:84)
>>>>>> at org.apache.spark.sql.catalyst.ScalaReflection$class.construc
>>>>>> tParams(ScalaReflection.scala:811)
>>>>>> at org.apache.spark.sql.catalyst.ScalaReflection$.constructPara
>>>>>> ms(ScalaReflection.scala:39)
>>>>>> at org.apache.spark.sql.catalyst.ScalaReflection$class.getConst
>>>>>> ructorParameters(ScalaReflection.scala:800)
>>>>>> at org.apache.spark.sql.catalyst.ScalaReflection$.getConstructo
>>>>>> rParameters(ScalaReflection.scala:39)
>>>>>> at org.apache.spark.sql.catalyst.ScalaReflection$.org$apache$sp
>>>>>> ark$sql$catalyst$ScalaReflection$$serializerFor(ScalaReflect
>>>>>> ion.scala:582)
>>>>>> at org.apache.spark.sql.catalyst.ScalaReflection$.org$apache$sp
>>>>>> ark$sql$catalyst$ScalaReflection$$serializerFor(ScalaReflect
>>>>>> ion.scala:460)
>>>>>> at org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$9.app
>>>>>> ly(ScalaReflection.scala:592)
>>>>>> at org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$9.app
>>>>>> ly(ScalaReflection.scala:583)
>>>>>> at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(Tr
>>>>>> aversableLike.scala:252)
>>>>>> at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(Tr
>>>>>> aversableLike.scala:252)
>>>>>> at scala.collection.immutable.List.foreach(List.scala:381)
>>>>>> at scala.collection.TraversableLike$class.flatMap(TraversableLi
>>>>>> ke.scala:252)
>>>>>> at scala.collection.immutable.List.flatMap(List.scala:344)
>>>>>> at org.apache.spark.sql.catalyst.ScalaReflection$.org$apache$sp
>>>>>> ark$sql$catalyst$ScalaReflection$$serializerFor(ScalaReflect
>>>>>> ion.scala:583)
>>>>>> at org.apache.spark.sql.catalyst.ScalaReflection$.serializerFor
>>>>>> (ScalaReflection.scala:425)
>>>>>> at org.apache.spark.sql.catalyst.encoders.ExpressionEncoder$.ap
>>>>>> ply(ExpressionEncoder.scala:61)
>>>>>> at org.apache.spark.sql.Encoders$.product(Encoders.scala:274)
>>>>>> at org.apache.spark.sql.SQLImplicits.newProductEncoder(SQLImpli
>>>>>> cits.scala:47)
>>>>>> at PersonConsumer$.main(PersonConsumer.scala:33)
>>>>>> at PersonConsumer.main(PersonConsumer.scala)
>>>>>> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
>>>>>> at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAcce
>>>>>> ssorImpl.java:62)
>>>>>> at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMe
>>>>>> thodAccessorImpl.java:43)
>>>>>> at java.lang.reflect.Method.invoke(Method.java:498)
>>>>>> at com.intellij.rt.execution.application.AppMain.main(AppMain.j
>>>>>> ava:147)
>>>>>>
>>>>>> The following is my code ...
>>>>>>
>>>>>> object PersonConsumer {
>>>>>> import org.apache.spark.rdd.RDD
>>>>>> import com.trueaccord.scalapb.spark._
>>>>>> import org.apache.spark.sql.{SQLContext, SparkSession}
>>>>>> import com.example.protos.demo._
>>>>>>
>>>>>> def main(args : Array[String]) {
>>>>>>
>>>>>> def parseLine(s: String): Person =
>>>>>> Person.parseFrom(
>>>>>> org.apache.commons.codec.binary.Base64.decodeBase64(s))
>>>>>>
>>>>>> val spark = SparkSession.builder.
>>>>>> master("local")
>>>>>> .appName("spark session example")
>>>>>> .getOrCreate()
>>>>>>
>>>>>> import spark.implicits._
>>>>>>
>>>>>> val ds1 =
>>>>>> spark.readStream.format("kafka").option("kafka.bootstrap.servers","localhost:9092").option("subscribe","person").load()
>>>>>>
>>>>>> val ds2 = ds1.selectExpr("CAST(value AS STRING)").as[String]
>>>>>>
>>>>>> val ds3 = ds2.map(str =>
>>>>>> parseLine(str)).createOrReplaceTempView("persons")
>>>>>>
>>>>>> val ds4 = spark.sqlContext.sql("select name from persons")
>>>>>>
>>>>>> val query = ds4.writeStream
>>>>>> .outputMode("append")
>>>>>> .format("console")
>>>>>> .start()
>>>>>> query.awaitTermination()
>>>>>> }
>>>>>> }
>>>>>>
>>>>>>
>>>>>
>>>>
>>>
>>
>
