"Neither Java nor Kyro works for class: class com.twitter.scalding.typed.FlatMappedFn"

[Sasha Ovsankin]

Hi, I am using Scalding 0.9rc1. 

The code looks like:

 val bar: TypedPipe[Bar]= ....
 val foo: TypedPipe[Foo]= bar.filter(e => "A".equals(e.getType) &&
                                          "B".equals(e. getSubtype)).flatMap

  { e =>
      e.getSomeList.asScala.filter { x => "X".equals(d.getType.toString) }
        .map { x=> new Foo(x.getA, x.getB) }
  }
  foo.write(PackedAvroSource[Foo]("myPath"))

Foo is a Java class that was generated off of Avro Schema.

Here is the exception info:

! Exception in thread "main" cascading.flow.planner.PlannerException: could not build flow from assembly: [Neither Java nor Kyro works for class: class com.twitter.scalding.typed.FlatMappedFn instance: <function1>

! export CHILL_EXTERNALIZER_DEBUG=true to see both stack traces]

! at cascading.flow.planner.FlowPlanner.handleExceptionDuringPlanning(FlowPlanner.java:533)

! at cascading.flow.hadoop.planner.HadoopPlanner.buildFlow(HadoopPlanner.java:237)

! at cascading.flow.FlowConnector.connect(FlowConnector.java:454)

! at com.twitter.scalding.Job.buildFlow(Job.scala:205)

! at com.twitter.scalding.Job.runFlow(Job.scala:226)

! at com.twitter.scalding.Job.run(Job.scala:232)

! at com.twitter.scalding.Tool.start$1(Tool.scala:114)

! at com.twitter.scalding.Tool.run(Tool.scala:132)

! at com.twitter.scalding.Tool.run(Tool.scala:70)

! at org.apache.hadoop.util.ToolRunner.run(ToolRunner.java:65)

This code worked in a slightly different context (I was sing the "foo" in further expressions)

Thanks,

-- Sasha

[Oscar Boykin]

So, in your code, "d" is a variable outside the function. This causes a closure to be built in your job.

Two quick hacks:

1) make all the vals lazy in the Job.

2) make all the TypedPipes @transient.

Did you turn on the environmental variable that it suggests in the stack trace? CHILL_....  That will tell you exactly what part of the serialization fails. Also see the scalding wiki FAQ on debugging serialization errors.

You may need to write/add serializers for items you need.

--
You received this message because you are subscribed to the Google Groups "cascading-user" group.
To unsubscribe from this group and stop receiving emails from it, send an email to cascading-use...@googlegroups.com.
To post to this group, send email to cascadi...@googlegroups.com.
Visit this group at http://groups.google.com/group/cascading-user.
For more options, visit https://groups.google.com/groups/opt_out.

--

Oscar Boykin :: @posco :: http://twitter.com/posco

[Christopher Severs]

Kryo has had trouble with certain Avro records in the past. Until I have a good Kryo based solution there is a Hadoop Avro SpecificRecord serializer in cascading.avro that you can put in front of Kryo in your job. I don't know that that's the problem you're running in to but it may be worth a try.

[Oscar Boykin]

We can add a chill-avro to:

https://github.com/twitter/chill/

and we can pull that in pretty easy. I guess we just need to add a kryo.Serializer for Avro.

[Christopher Severs]

Yeah that's the best solution. I'll take a stab at it.

[Sasha Ovsankin]

If this is of any help, the object it tries to serialize when it fails is an instance is com.twitter.scalding.FlatMapFunction[decl:0]. We have reduced the problem down to this code:

  val foo: TypedPipe[(Int, Long)]= bar.flatMap { e =>

    e.getSomeList.asScala.map { x=> (x.getA, x.getB) }

  }

I am not sure how all this is related to Avro (except for 'bar') which is an Avro source

You received this message because you are subscribed to a topic in the Google Groups "cascading-user" group.
To unsubscribe from this topic, visit https://groups.google.com/d/topic/cascading-user/CNFiAmkBMUQ/unsubscribe.
To unsubscribe from this group and all its topics, send an email to cascading-use...@googlegroups.com.

[Oscar Boykin]

Man. I sure do wish I could solve something other than serialization problems.

I'd love to retire from these bugs, and just remember fondly, and now the young guys come along, who are convinced I was a fool or incompetent or both.

I'd live on a small island, by the warm, calm sea in a shipping container, with family and dog.

And serialization would be a solved problem, like getting nand gates to work or something. Just something routine, that happened without much notice.

This is my dream....

Sorry our collective fiction of reliable, effort-free serialization is failing you right now. You didn't mention if you tried the lazy (for things used in functions) and/or transient (for pipes) hack worked. I guess not, but please do share your results.

[Sasha Ovsankin]

Hi Oscar --

I heeded to your and your code's repeated calls to CHILL_EXTERNALIZER_DEBUG and here is what I found:

trait SourceUtils
{
  final private val log: Logger = LoggerFactory.getLogger(getClass)
  ...
}
class Discussions(args: Args) extends LiJob(args) with SourceUtils
{
  final private val log = LoggerFactory.getLogger(getClass)
  ...
}

The above code throws the exception in the enclosed log. Marking both definitions @transient makes the exception go away. 

It doesn't look like it was related to FlatMap. The matters were further complicated by IntelliJ's Scala compiler stubborn reluctance to recompile the changed code.

So repeat after me, children: "@transient final private val log".

Guess you can book your tickets to New Caledonia now.

Thanks!

-- Sasha

[Oscar Boykin]

Glad to see you had a solution.

I added the issue about Loggers to the FAQ:

https://github.com/twitter/scalding/wiki/Frequently-asked-questions#q-im-getting-a-notserializableexception-on-hadoop-job-submission