This is resolved by the first approach I mentioned.
Thanks Team
On Mon, 30 Aug 2021 at 12:35, tarun joshi <1985.ta...@gmail.com> wrote:
> An update on this , I see that `IndexedRecord` is part of Avro Library.
> Please correct me If I am wrong in assuming that the "Pojo's generated by
> Avro POJO generator must be implementing IndexedRecord interface" It seems
> either
>
>
> - I should be parsing Stringified Json from AWS Kinesis directly into
> Avro.
> - *Or* convert those GSON parsed POJOs into Avro compatible POJOs at
> Stream-time.
>
> Please let me know if anyone has a better way to do this.
>
>
> On Mon, 30 Aug 2021 at 10:13, tarun joshi <1985.ta...@gmail.com> wrote:
>
>> Hey all,
>> I am trying to write a simple pipeline to read
>>
>> Read Stringified JSON from Kinesis -> parsed to POJO -> converted to Avro
>> -> for the purpose of writing Parquet files to AWS S3.
>>
>> 1) This is my SimpleMapper
>>
>> public class SimpleMapper extends RichMapFunction<String, GenericRecord> {
>> private static final GsonBuilder gsonBuilder =
>> new
>> GsonBuilder().excludeFieldsWithoutExposeAnnotation().setPrettyPrinting();
>>
>> private static final Gson gson = gsonBuilder.create();
>> private static final Schema schema =
>> ReflectData.get().getSchema(Response.class);
>>
>> @Override
>> public GenericRecord map(String s) throws Exception {
>>
>> Response response = gson.fromJson(s, Response.class);
>> GenericData.Record record = new GenericData.Record(schema);
>> record.put(0, response);
>>
>> return record;
>> }
>>
>> 2) This is my Job Definition
>>
>> public class ClickStreamPipeline implements Serializable {
>>
>> private static Schema schema =
>> ReflectData.get().getSchema(Response.class);
>>
>> public static void main(String args[]) throws Exception {
>> final MultipleParameterTool params =
>> MultipleParameterTool.fromArgs(args);
>> StreamExecutionEnvironment env =
>> getStreamExecutionEnvironment(params);
>>
>>
>> FlinkKinesisConsumer<String> kinesisConsumer =
>> new FlinkKinesisConsumer<>(
>> "web-clickstream", new SimpleStringSchema(),
>> getKafkaConsumerProperties());
>>
>> final StreamingFileSink<GenericRecord> streamingFileSink =
>> StreamingFileSink.forBulkFormat(
>> new
>> Path("s3://data-ingestion-pipeline/flink_pipeline/"),
>> ParquetAvroWriters.forGenericRecord(schema))
>> .withRollingPolicy(OnCheckpointRollingPolicy.build())
>> .build();
>>
>> env.addSource(kinesisConsumer)
>> .map(new SimpleMapper())
>> .returns(new GenericRecordAvroTypeInfo(schema))
>> .addSink(streamingFileSink);
>>
>> env.execute("Read files in streaming fashion");
>> }
>>
>> private static StreamExecutionEnvironment getStreamExecutionEnvironment(
>> MultipleParameterTool params) throws ClassNotFoundException {
>> StreamExecutionEnvironment env =
>> StreamExecutionEnvironment.getExecutionEnvironment();
>> env.getConfig().setGlobalJobParameters(params);
>>
>> Class<?> unmodColl =
>> Class.forName("java.util.Collections$UnmodifiableCollection");
>> env.getConfig()
>> .addDefaultKryoSerializer(unmodColl,
>> UnmodifiableCollectionsSerializer.class);
>> env.enableCheckpointing(60_000L);
>>
>> return env;
>> }
>>
>>
>>
>> The issue I am facing is failiing to serialize the Avro GenericRecord
>> wrapped message
>>
>> - When I used a GenericRecordAvroTypeInfo(schema); to force use my
>> Avro as preferred Serializer , I am getting the error below
>>
>>
>> * java.lang.ClassCastException: class <my fully qualified
>> POJO> cannot be cast to class org.apache.avro.generic.IndexedRecord*
>>
>>
>>
>> - If I don't use the GenericRecordAvroTypeInfo and try to register my
>> pojo with KryoSerializer , the serialization fails with NPE somewhere in
>> my
>> Schema class.Do I need to implement/register a proper Avro serializer with
>> flink config?
>>
>> Thanks for the help!
>>
>
