Hi Porritt, Based on the exception stack trace you provided, it seems the exception occurs when initializing Avro schema. You did not give the definition of the MyAvroSchema Class, so I'd to suggest you :
1. make sure the file path "file:///home/myuser/test.avro" exists in your tm node which run your source task. 2. here is the flink-avro connector documentation[1] you can refer 3. there are many test case such as this[2], you can use to test your program, just need a little change [1]: https://ci.apache.org/projects/flink/flink-docs-release-1.5/dev/batch/connectors.html#avro-support-in-flink [2]: https://github.com/apache/flink/blob/master/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroInputFormatTypeExtractionTest.java --- thanks. vino. 2018-07-16 20:22 GMT+08:00 Porritt, James <james.porr...@uk.mlp.com>: > I’ve been trying to use the following code: > > > > ExecutionEnvironment env = ExecutionEnvironment. > getExecutionEnvironment(); > > Path path = new Path("file:///home/myuser/test.avro"); > > AvroInputFormat<MyAvroSchema> my_format = new > AvroInputFormat<>(path, MyAvroSchema.class); > > DataSet<MyAvroSchema> my_input = env.createInput(my_format); > > my_input.print(); > > env.execute(); > > > > to utilise this avro schema: > > > > { > > "type" : "record", > > "name" : "MyAvroSchema", > > "fields" : [ { > > "name" : "a", > > "type" : [ "null", "int" ] > > }, { > > "name" : "b", > > "type" : [ "null", "string" ] > > }] > > } > > > > I created the MyAvroSchema class from this schema using avro tools. I also > converted the following JSON into a compatible avro stored in > file:///home/myuser/test.avro > > > > {"a":{"int":123},"b":{"string":"hello"}} > > > > When I try and run this however I get: > > > > 2018-07-16 12:59:26,761 INFO org.apache.flink.runtime. > executiongraph.ExecutionGraph - DataSource (at > createInput(ExecutionEnvironment.java:548) > (org.apache.flink.formats.avro.AvroInputFormat) > > ) (1/1) (302878b522f420f6b7866de4f32fcbd6) switched from RUNNING to > FAILED. > > org.apache.avro.AvroRuntimeException: avro.shaded.com.google.common. > util.concurrent.UncheckedExecutionException: > java.lang.NullPointerException > > at org.apache.avro.specific.SpecificData.getSchema( > SpecificData.java:227) > > at org.apache.avro.specific.SpecificDatumReader.<init>( > SpecificDatumReader.java:37) > > at org.apache.flink.formats.avro.AvroInputFormat.initReader( > AvroInputFormat.java:122) > > at org.apache.flink.formats.avro.AvroInputFormat.open( > AvroInputFormat.java:111) > > at org.apache.flink.formats.avro.AvroInputFormat.open( > AvroInputFormat.java:54) > > at org.apache.flink.runtime.operators.DataSourceTask. > invoke(DataSourceTask.java:170) > > at org.apache.flink.runtime.taskmanager.Task.run(Task.java:703) > > at java.lang.Thread.run(Thread.java:748) > > Caused by: > avro.shaded.com.google.common.util.concurrent.UncheckedExecutionException: > java.lang.NullPointerException > > at avro.shaded.com.google.common.cache.LocalCache$Segment.get( > LocalCache.java:2234) > > at avro.shaded.com.google.common.cache.LocalCache.get( > LocalCache.java:3965) > > at avro.shaded.com.google.common.cache.LocalCache.getOrLoad( > LocalCache.java:3969) > > at avro.shaded.com.google.common.cache.LocalCache$ > LocalManualCache.get(LocalCache.java:4829) > > at org.apache.avro.specific.SpecificData.getSchema( > SpecificData.java:225) > > ... 7 more > > Caused by: java.lang.NullPointerException > > at java.lang.String.replace(String.java:2239) > > at org.apache.avro.specific.SpecificData.createSchema( > SpecificData.java:281) > > at org.apache.avro.specific.SpecificData$2.load( > SpecificData.java:218) > > at org.apache.avro.specific.SpecificData$2.load( > SpecificData.java:215) > > at avro.shaded.com.google.common.cache.LocalCache$ > LoadingValueReference.loadFuture(LocalCache.java:3568) > > at avro.shaded.com.google.common.cache.LocalCache$Segment. > loadSync(LocalCache.java:2350) > > at avro.shaded.com.google.common.cache.LocalCache$Segment. > lockedGetOrLoad(LocalCache.java:2313) > > at avro.shaded.com.google.common.cache.LocalCache$Segment.get( > LocalCache.java:2228) > > ... 11 more > > > > Can anyone suggest what might be causing the NullPointerException? I’m > using flink-1.5.0 and avro-tools-1.8.2 > > > ###################################################################### > The information contained in this communication is confidential and > intended only for the individual(s) named above. If you are not a named > addressee, please notify the sender immediately and delete this email > from your system and do not disclose the email or any part of it to any > person. The views expressed in this email are the views of the author > and do not necessarily represent the views of Millennium Capital Partners > LLP (MCP LLP) or any of its affiliates. Outgoing and incoming electronic > communications of MCP LLP and its affiliates, including telephone > communications, may be electronically archived and subject to review > and/or disclosure to someone other than the recipient. MCP LLP is > authorized and regulated by the Financial Conduct Authority. Millennium > Capital Partners LLP is a limited liability partnership registered in > England & Wales with number OC312897 and with its registered office at > 50 Berkeley Street, London, W1J 8HD > <https://maps.google.com/?q=50+Berkeley+Street,+London,+W1J+8HD&entry=gmail&source=g> > . > ###################################################################### > >
