title: How to use a custom Coder in a PCollection<KV<String, B>>?
body: <p>I'm trying to use a custom <code>Coder</code> so that I can do some transforms, but I'm having trouble getting the <code>PCollection</code> to use my custom coder, and I <em>suspect</em> (???) it's because it's wrapped in a <code>KV</code>. Specifically:</p>
  
  <pre><code>Pipeline p = Pipeline.create ...
  p.getCoderRegistry().registerCoder(MyClass.class, MyClassCoder.class);
  
  ...
  
  PCollection&lt;String&gt; input = ...
  PCollection&lt;KV&lt;String, MyClass&gt;&gt; t = input.apply(new ToKVTransform());
  </code></pre>
  
  <p>When I try to run something like this, I get a java.lang.ClassCastException and a stacktrace that includes a <code>SerializableCoder</code> instead of <code>MyClassCoder</code> like I would expect.</p>
  
  <pre><code>[error]     at com.google.cloud.dataflow.sdk.coders.SerializableCoder.decode(SerializableCoder.java:133)
  [error]     at com.google.cloud.dataflow.sdk.coders.SerializableCoder.decode(SerializableCoder.java:50)
  [error]     at com.google.cloud.dataflow.sdk.coders.KvCoder.decode(KvCoder.java:95)
  [error]     at com.google.cloud.dataflow.sdk.coders.KvCoder.decode(KvCoder.java:42)
  </code></pre>
  
  <p>I see that the answer to another, somewhat related question (<a href="https://stackoverflow.com/questions/29131859/using-textio-write-with-a-complicated-pcollection-type-in-google-cloud-dataflow">Using TextIO.Write with a complicated PCollection type in Google Cloud Dataflow</a>) says to map everything to strings, and use that to pass stuff around PCollections. Is that <em>really</em> the recommended way??</p>
  
  <p>(Note: the actual code is in Scala, but I'm pretty sure it's not a Scala &lt;=> Java issue so I've translated it into Java here.)</p>
  
  <p>Update to include Scala code and more background:</p>
  
  <p>So this is the actual exception itself (should have included this at the beginning):</p>
  
  <pre><code>java.lang.ClassCastException: cannot assign instance of scala.collection.immutable.HashMap$SerializationProxy to field com.example.schema.Schema.keyTypes of type scala.collection.immutable.Map in instance of com.example.schema.Schema
  </code></pre>
  
  <p>Where <code>com.example.schema.Schema</code> is:</p>
  
  <pre><code>case class Schema(id: String, keyTypes: Map[String, Type])
  </code></pre>
  
  <p>And lastly, the <code>SchemaCoder</code> is:</p>
  
  <pre><code>class SchemaCoder extends com.google.cloud.dataflow.sdk.coders.CustomCoder[Schema] {
    def decode(inputStream: InputStream, context: Context): Schema = {
      val ois = new ObjectInputStream(inputStream)
      val id: String = ois.readObject().asInstanceOf[String]
      val javaMap: java.util.Map[String, Type] = ois.readObject().asInstanceOf[java.util.Map[String, Type]]
      ois.close()
  
      Schema(id, javaMap.asScala.toMap)
    }
  
    def encode(schema: Schema, outputStream: OutputStream, context: Context): Unit = {
      val baos = new ByteArrayOutputStream()
      val oos = new ObjectOutputStream(baos)
      oos.writeObject(schema.id)
      val javaMap: java.util.Map[String, Type] = schema.keyTypes.asJava
      oos.writeObject(javaMap)
      oos.close()
  
      val encoded = new String(Base64.encodeBase64(baos.toByteArray()))
      outputStream.write(encoded.getBytes())
    }
  }
  </code></pre>
  
  <p>====</p>
  
  <p>Edit2: And here's what <code>ToKVTransform</code> actually looks like:</p>
  
  <pre><code>class SchemaExtractorTransform extends PTransform[PCollection[String], PCollection[Schema]] {
    class InferSchemaFromStringWithKeyFn extends DoFn[String, KV[String, Schema]] {
      override def processElement(c: DoFn[String, KV[String, Schema]]#ProcessContext): Unit = {
        val line = c.element()
        inferSchemaFromString(line)
      }
    }
  
    class GetFirstFn extends DoFn[KV[String, java.lang.Iterable[Schema]], Schema] {
      override def processElement(c: DoFn[KV[String, java.lang.Iterable[Schema]], Schema]#ProcessContext): Unit = {
        val idAndSchemas: KV[String, java.lang.Iterable[Schema]] = c.element()
        val it: java.util.Iterator[Schema] = idAndSchemas.getValue().iterator()
        c.output(it.next())
      }
    }
  
    override def apply(inputLines: PCollection[String]): PCollection[Schema] = {
      val schemasWithKey: PCollection[KV[String, Schema]] = inputLines.apply(
        ParDo.named("InferSchemas").of(new InferSchemaFromStringWithKeyFn())
      )
  
      val keyed: PCollection[KV[String, java.lang.Iterable[Schema]]] = schemasWithKey.apply(
        GroupByKey.create()
      )
  
      val schemasOnly: PCollection[Schema] = keyed.apply(
        ParDo.named("GetFirst").of(new GetFirstFn())
      )
  
      schemasOnly
    }
  }
  </code></pre>