EncoderHelpers { public static <T> Encoder<T> fromBeamCoder(Coder<T> coder) { Class<? super T> clazz = coder.getEncodedTypeDescriptor().getRawType(); ClassTag<T> classTag = ClassTag$.MODULE$.apply(clazz); List<Expression> serializers = Collections.singletonList( new EncodeUsingBeamCoder<>(new BoundReference(0, new ObjectType(clazz), true), coder)); return new ExpressionEncoder<>( SchemaHelpers.binarySchema(), false, JavaConversions.collectionAsScalaIterable(serializers).toSeq(), new DecodeUsingBeamCoder<>( new Cast(new GetColumnByOrdinal(0, BinaryType), BinaryType), classTag, coder), classTag); }  static Encoder<T> fromBeamCoder(Coder<T> coder);  }
@Test public void beamCoderToSparkEncoderTest() { SparkSession sparkSession = SparkSession.builder() .appName("beamCoderToSparkEncoderTest") .master("local[4]") .getOrCreate(); List<Integer> data = Arrays.asList(1, 2, 3); Dataset<Integer> dataset = sparkSession.createDataset(data, EncoderHelpers.fromBeamCoder(VarIntCoder.of())); assertEquals(data, dataset.collectAsList()); }